Skip to content

[processor] Set ReceiveInterval to be 0 when we receive less than requested #277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[processor] Set ReceiveInterval to be 0 when we receive less than requested #277

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions v2/processor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,9 @@ type ProcessorOptions struct {
// MaxReceiveCount is the maximum number of messages to receive at a time. This value is passed to the receiver.
MaxReceiveCount int

// ReceiveInterval is the interval between each receive call.
// ReceiveInterval is the interval between each receive call when there is enough available concurrency.
// If we receive less than the number of messages requested, the next receive is called immediately
// without waiting for the ReceiveInterval.
ReceiveInterval *time.Duration

// StartMaxAttempt is the maximum number of attempts to start the processor.
Expand Down Expand Up @@ -213,9 +215,10 @@ func (p *Processor) start(ctx context.Context, receiverEx *ReceiverEx) error {
for _, msg := range messages {
p.process(ctx, receiverEx, msg)
}
receiveInterval := *p.options.ReceiveInterval
for ctx.Err() == nil {
select {
case <-time.After(*p.options.ReceiveInterval):
case <-time.After(receiveInterval):
maxMessages := min(p.options.MaxReceiveCount, p.options.MaxConcurrency-len(p.concurrencyTokens))
if ctx.Err() != nil || maxMessages == 0 {
break
Expand All @@ -229,6 +232,12 @@ func (p *Processor) start(ctx context.Context, receiverEx *ReceiverEx) error {
for _, msg := range messages {
p.process(ctx, receiverEx, msg)
}
// if we received less than maxMessages, the next ReceiveMessages() is called immediately
if len(messages) < maxMessages {
receiveInterval = 0
} else {
receiveInterval = *p.options.ReceiveInterval
}
case <-ctx.Done():
logger.Info(fmt.Sprintf("context done, stop receiving from processor %s", receiverName))
break
Expand Down
Loading