@@ -297,6 +297,73 @@ func (t PullThresholdBytes) configureMessages(opts *consumeOpts) error {
297297 return nil
298298}
299299
300+ // PullMinPending sets the minimum number of messages that should be pending for
301+ // a consumer with PriorityPolicyOverflow to be considered for delivery.
302+ // If provided, PullPriorityGroup must be set as well and the consumer has to have
303+ // PriorityPolicy set to PriorityPolicyOverflow.
304+ //
305+ // PullMinPending implements both PullConsumeOpt and PullMessagesOpt, allowing
306+ // it to configure Consumer.Consume and Consumer.Messages.
307+ type PullMinPending int
308+
309+ func (min PullMinPending ) configureConsume (opts * consumeOpts ) error {
310+ if min < 1 {
311+ return fmt .Errorf ("%w: min pending should be more than 0" , ErrInvalidOption )
312+ }
313+ opts .MinPending = int64 (min )
314+ return nil
315+ }
316+
317+ func (min PullMinPending ) configureMessages (opts * consumeOpts ) error {
318+ if min < 1 {
319+ return fmt .Errorf ("%w: min pending should be more than 0" , ErrInvalidOption )
320+ }
321+ opts .MinPending = int64 (min )
322+ return nil
323+ }
324+
325+ // PullMinAckPending sets the minimum number of pending acks that should be
326+ // present for a consumer with PriorityPolicyOverflow to be considered for
327+ // delivery. If provided, PullPriorityGroup must be set as well and the consumer
328+ // has to have PriorityPolicy set to PriorityPolicyOverflow.
329+ //
330+ // PullMinAckPending implements both PullConsumeOpt and PullMessagesOpt, allowing
331+ // it to configure Consumer.Consume and Consumer.Messages.
332+ type PullMinAckPending int
333+
334+ func (min PullMinAckPending ) configureConsume (opts * consumeOpts ) error {
335+ if min < 1 {
336+ return fmt .Errorf ("%w: min pending should be more than 0" , ErrInvalidOption )
337+ }
338+ opts .MinAckPending = int64 (min )
339+ return nil
340+ }
341+
342+ func (min PullMinAckPending ) configureMessages (opts * consumeOpts ) error {
343+ if min < 1 {
344+ return fmt .Errorf ("%w: min pending should be more than 0" , ErrInvalidOption )
345+ }
346+ opts .MinAckPending = int64 (min )
347+ return nil
348+ }
349+
350+ // PullPriorityGroup sets the priority group for a consumer.
351+ // It has to match one of the priority groups set on the consumer.
352+ //
353+ // PullPriorityGroup implements both PullConsumeOpt and PullMessagesOpt, allowing
354+ // it to configure Consumer.Consume and Consumer.Messages.
355+ type PullPriorityGroup string
356+
357+ func (g PullPriorityGroup ) configureConsume (opts * consumeOpts ) error {
358+ opts .Group = string (g )
359+ return nil
360+ }
361+
362+ func (g PullPriorityGroup ) configureMessages (opts * consumeOpts ) error {
363+ opts .Group = string (g )
364+ return nil
365+ }
366+
300367// PullHeartbeat sets the idle heartbeat duration for a pull subscription
301368// If a client does not receive a heartbeat message from a stream for more
302369// than the idle heartbeat setting, the subscription will be removed
@@ -368,6 +435,43 @@ func WithMessagesErrOnMissingHeartbeat(hbErr bool) PullMessagesOpt {
368435 })
369436}
370437
438+ // FetchMinPending sets the minimum number of messages that should be pending for
439+ // a consumer with PriorityPolicyOverflow to be considered for delivery.
440+ // If provided, FetchPriorityGroup must be set as well and the consumer has to have
441+ // PriorityPolicy set to PriorityPolicyOverflow.
442+ func FetchMinPending (min int64 ) FetchOpt {
443+ return func (req * pullRequest ) error {
444+ if min < 1 {
445+ return fmt .Errorf ("%w: min pending should be more than 0" , ErrInvalidOption )
446+ }
447+ req .MinPending = min
448+ return nil
449+ }
450+ }
451+
452+ // FetchMinAckPending sets the minimum number of pending acks that should be
453+ // present for a consumer with PriorityPolicyOverflow to be considered for
454+ // delivery. If provided, FetchPriorityGroup must be set as well and the consumer
455+ // has to have PriorityPolicy set to PriorityPolicyOverflow.
456+ func FetchMinAckPending (min int64 ) FetchOpt {
457+ return func (req * pullRequest ) error {
458+ if min < 1 {
459+ return fmt .Errorf ("%w: min ack pending should be more than 0" , ErrInvalidOption )
460+ }
461+ req .MinAckPending = min
462+ return nil
463+ }
464+ }
465+
466+ // FetchPriorityGroup sets the priority group for a consumer.
467+ // It has to match one of the priority groups set on the consumer.
468+ func FetchPriorityGroup (group string ) FetchOpt {
469+ return func (req * pullRequest ) error {
470+ req .Group = group
471+ return nil
472+ }
473+ }
474+
371475// FetchMaxWait sets custom timeout for fetching predefined batch of messages.
372476//
373477// If not provided, a default of 30 seconds will be used.
0 commit comments