-
Notifications
You must be signed in to change notification settings - Fork 662
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.
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
Changes to default pending and default error handler. #607
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,4 +1,4 @@ | ||
// Copyright 2013-2019 The NATS Authors | ||
// Copyright 2013-2020 The NATS Authors | ||
// Licensed under the Apache License, Version 2.0 (the "License"); | ||
// you may not use this file except in compliance with the License. | ||
// You may obtain a copy of the License at | ||
|
@@ -417,6 +417,9 @@ func TestSlowSubscriber(t *testing.T) { | |
nc := NewDefaultConnection(t) | ||
defer nc.Close() | ||
|
||
// Override default handler for test. | ||
nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}) | ||
|
||
sub, _ := nc.SubscribeSync("foo") | ||
sub.SetPendingLimits(100, 1024) | ||
|
||
|
@@ -444,6 +447,9 @@ func TestSlowChanSubscriber(t *testing.T) { | |
nc := NewDefaultConnection(t) | ||
defer nc.Close() | ||
|
||
// Override default handler for test. | ||
nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}) | ||
|
||
ch := make(chan *nats.Msg, 64) | ||
sub, _ := nc.ChanSubscribe("foo", ch) | ||
sub.SetPendingLimits(100, 1024) | ||
|
@@ -467,6 +473,9 @@ func TestSlowAsyncSubscriber(t *testing.T) { | |
nc := NewDefaultConnection(t) | ||
defer nc.Close() | ||
|
||
// Override default handler for test. | ||
nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}) | ||
|
||
bch := make(chan bool) | ||
|
||
sub, _ := nc.Subscribe("foo", func(m *nats.Msg) { | ||
|
@@ -895,7 +904,8 @@ func TestChanSubscriberPendingLimits(t *testing.T) { | |
// There was a defect that prevented to receive more than | ||
// the default pending message limit. Trying to send more | ||
// than this limit. | ||
total := nats.DefaultSubPendingMsgsLimit + 100 | ||
pending := 1000 | ||
total := pending + 100 | ||
|
||
for typeSubs := 0; typeSubs < 3; typeSubs++ { | ||
|
||
|
@@ -908,10 +918,19 @@ func TestChanSubscriberPendingLimits(t *testing.T) { | |
switch typeSubs { | ||
case 0: | ||
sub, err = nc.ChanSubscribe("foo", ch) | ||
if err := sub.SetPendingLimits(pending, -1); err == nil { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I thought limits did not apply to Channel subscribe, because the limit is really whatever the user sets for the provided channel? There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The test expects an error since its checking for |
||
t.Fatalf("Expected an error setting pending limits") | ||
} | ||
case 1: | ||
sub, err = nc.ChanQueueSubscribe("foo", "bar", ch) | ||
if err := sub.SetPendingLimits(pending, -1); err == nil { | ||
t.Fatalf("Expected an error setting pending limits") | ||
} | ||
case 2: | ||
sub, err = nc.QueueSubscribeSyncWithChan("foo", "bar", ch) | ||
if err := sub.SetPendingLimits(pending, -1); err == nil { | ||
t.Fatalf("Expected an error setting pending limits") | ||
} | ||
} | ||
if err != nil { | ||
t.Fatalf("Unexpected error on subscribe: %v", err) | ||
|
@@ -1020,6 +1039,9 @@ func TestUnsubscribeChanOnSubscriber(t *testing.T) { | |
nc := NewDefaultConnection(t) | ||
defer nc.Close() | ||
|
||
// Override default handler for test. | ||
nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}) | ||
|
||
// Create our own channel. | ||
ch := make(chan *nats.Msg, 8) | ||
sub, _ := nc.ChanSubscribe("foo", ch) | ||
|
@@ -1285,6 +1307,9 @@ func TestSetPendingLimits(t *testing.T) { | |
nc := NewDefaultConnection(t) | ||
defer nc.Close() | ||
|
||
// Override default handler for test. | ||
nc.SetErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, _ error) {}) | ||
|
||
payload := []byte("hello") | ||
payloadLen := len(payload) | ||
toSend := 100 | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This function was protected by the subsMu so that remove could not happen while processing a message. I believe this is the reason we now can get the panic:
You need to check for
sub.closed
line 2466, after lock the sub. If closed, return.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
By the way, for this test: TestUnsubscribeChanOnSubscriber, I also see the slow consumer error reported in some cases, so may have to set dummy error handler there too.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok will take a deeper look.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
ok fixed.