Skip to content
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

Jump to bottom

Allow setting consumer replicas though options #1019

Merged
merged 8 commits into from Jul 29, 2022
Merged

Allow setting consumer replicas though options #1019

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
5240d8e
Allow setting consumer replicas though options
Jul 22, 2022
c5c2b8e
Run go fmt
Jul 22, 2022
f732c53
Add test for consumer replica option
Jul 28, 2022
122e3a1
Remove redundant return statement
Jul 28, 2022
1d9feeb
Test consumer replica option with clustered JetStream
Jul 29, 2022
aff87fc
Merge remote-tracking branch 'upstream/main'
Jul 29, 2022
c50c8f9
Fix merge conflicts
Jul 29, 2022
dbc90ca
Run go fmt
Jul 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
5 changes: 5 additions & 0 deletions example_test.go
View file
Open in desktop
Expand Up @@ -625,6 +625,11 @@ func ExampleSubOpt() {
js.Subscribe("foo", func(msg *nats.Msg) {
fmt.Printf("Received a message: %s\n", string(msg.Data))
}, nats.ManualAck(), nats.MaxDeliver(2), nats.BackOff([]time.Duration{50 * time.Millisecond, 250 * time.Millisecond}))

// Set consumer replicas count for a durable while subscribing.
js.Subscribe("foo", func(msg *nats.Msg) {
fmt.Printf("Received a message: %s\n", string(msg.Data))
}, nats.Durable("FOO"), nats.ConsumerReplicas(1))
}

func ExampleMaxWait() {
Expand Down
11 changes: 11 additions & 0 deletions js.go
View file
Open in desktop
Expand Up @@ -2413,6 +2413,17 @@ func InactiveThreshold(threshold time.Duration) SubOpt {
})
}

// ConsumerReplicas sets the number of replica count for a consumer.
func ConsumerReplicas(replicas int) SubOpt {
return subOptFn(func(opts *subOpts) error {
if replicas < 1 {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think we can also consider setting adding && replicas <=5, as that is the upper limit.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is the current server limit, but since server defined and we could change would leave that out..

return fmt.Errorf("invalid ConsumerReplicas value (%v), needs to be greater than 0", replicas)
}
opts.cfg.Replicas = replicas
return nil
})
}

func (sub *Subscription) ConsumerInfo() (*ConsumerInfo, error) {
sub.mu.Lock()
// TODO(dlc) - Better way to mark especially if we attach.
Expand Down