@@ -22,6 +22,7 @@ type StreamCmdable interface {
2222 XGroupCreateConsumer (ctx context.Context , stream , group , consumer string ) * IntCmd
2323 XGroupDelConsumer (ctx context.Context , stream , group , consumer string ) * IntCmd
2424 XReadGroup (ctx context.Context , a * XReadGroupArgs ) * XStreamSliceCmd
25+ XReadGroupNoBlock (ctx context.Context , a * XReadGroupArgs ) * XStreamSliceCmd
2526 XAck (ctx context.Context , stream , group string , ids ... string ) * IntCmd
2627 XPending (ctx context.Context , stream , group string ) * XPendingCmd
2728 XPendingExt (ctx context.Context , a * XPendingExtArgs ) * XPendingExtCmd
@@ -221,6 +222,30 @@ type XReadGroupArgs struct {
221222 NoAck bool
222223}
223224
225+ func (c cmdable ) XReadGroupNoBlock (ctx context.Context , a * XReadGroupArgs ) * XStreamSliceCmd {
226+ args := make ([]interface {}, 0 , 10 + len (a .Streams ))
227+ args = append (args , "xreadgroup" , "group" , a .Group , a .Consumer )
228+
229+ keyPos := int8 (4 )
230+ if a .Count > 0 {
231+ args = append (args , "count" , a .Count )
232+ keyPos += 2
233+ }
234+ if a .NoAck {
235+ args = append (args , "noack" )
236+ keyPos ++
237+ }
238+ args = append (args , "streams" )
239+ keyPos ++
240+ for _ , s := range a .Streams {
241+ args = append (args , s )
242+ }
243+ cmd := NewXStreamSliceCmd (ctx , args ... )
244+ cmd .SetFirstKeyPos (keyPos )
245+ _ = c (ctx , cmd )
246+ return cmd
247+ }
248+
224249func (c cmdable ) XReadGroup (ctx context.Context , a * XReadGroupArgs ) * XStreamSliceCmd {
225250 args := make ([]interface {}, 0 , 10 + len (a .Streams ))
226251 args = append (args , "xreadgroup" , "group" , a .Group , a .Consumer )
0 commit comments