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.

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

Connection/Channel.NotifyClose: allow only buffered channel=1 #256

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion _examples/consumer/consumer.go
Expand Up @@ -96,7 +96,7 @@ func NewConsumer(amqpURI, exchange, exchangeType, queueName, key, ctag string) (
}

go func() {
Log.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
Log.Printf("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error, 1)))
}()

Log.Printf("got Connection, getting Channel")
Expand Down
6 changes: 6 additions & 0 deletions channel.go
Expand Up @@ -504,8 +504,14 @@ graceful close, no error will be sent.

In case of a non graceful close the error will be notified synchronously by the library
so that it will be necessary to consume the Channel from the caller in order to avoid deadlocks

The chan provided must be a buffered channel of size 1.
*/
func (ch *Channel) NotifyClose(c chan *Error) chan *Error {
if cap(c) != 1 {
panic("channel.NotifyClose expectes cap=1 buffered channel")
}

ch.notifyM.Lock()
defer ch.notifyM.Unlock()

Expand Down
79 changes: 77 additions & 2 deletions client_test.go
Expand Up @@ -632,7 +632,40 @@ func TestNotifyClosesReusedPublisherConfirmChan(t *testing.T) {
}
}

func TestNotifyClosesAllChansAfterConnectionClose(t *testing.T) {
func TestConnectionNotifyCloseAcceptsOnlyBufferedChannels(t *testing.T) {
rwc, srv := newSession(t)
t.Cleanup(func() { rwc.Close() })

go func() {
srv.connectionOpen()

srv.recv(0, &connectionClose{})
srv.send(0, &connectionCloseOk{})
}()

c, err := Open(rwc, defaultConfig())
if err != nil {
t.Fatalf("could not create connection: %v (%s)", c, err)
}

if err := c.Close(); err != nil {
t.Fatalf("could not close connection: %v (%s)", c, err)
}

defer func() {
_ = recover()
}()

select {
case <-c.NotifyClose(make(chan *Error)):
case <-time.After(time.Millisecond):
t.Errorf("expected to close NotifyClose chan after Connection.Close")
}

t.Errorf("connection.NotifyClose shouldn't accept unbuffered channels")
}

func TestChannelNotifyClosesAllChansAfterConnectionClose(t *testing.T) {
rwc, srv := newSession(t)

go func() {
Expand All @@ -657,8 +690,10 @@ func TestNotifyClosesAllChansAfterConnectionClose(t *testing.T) {
t.Fatalf("could not close connection: %v (%s)", c, err)
}

defer func() { _ = recover() }()

select {
case <-c.NotifyClose(make(chan *Error)):
case <-c.NotifyClose(make(chan *Error, 1)):
case <-time.After(time.Millisecond):
t.Errorf("expected to close NotifyClose chan after Connection.Close")
}
Expand All @@ -669,6 +704,46 @@ func TestNotifyClosesAllChansAfterConnectionClose(t *testing.T) {
t.Errorf("expected to close Connection.NotifyClose chan after Connection.Close")
}

t.Errorf("connection.NotifyClose shouldn't accept unbuffered channels")
}

func TestNotifyClosesAllChansAfterConnectionClose(t *testing.T) {
rwc, srv := newSession(t)

go func() {
srv.connectionOpen()
srv.channelOpen(1)

srv.recv(0, &connectionClose{})
srv.send(0, &connectionCloseOk{})
}()

c, err := Open(rwc, defaultConfig())
if err != nil {
t.Fatalf("could not create connection: %v (%s)", c, err)
}

ch, err := c.Channel()
if err != nil {
t.Fatalf("could not open channel: %v (%s)", ch, err)
}

if err := c.Close(); err != nil {
t.Fatalf("could not close connection: %v (%s)", c, err)
}

select {
case <-c.NotifyClose(make(chan *Error, 1)):
case <-time.After(time.Millisecond):
t.Errorf("expected to close NotifyClose chan after Connection.Close")
}

select {
case <-ch.NotifyClose(make(chan *Error, 1)):
case <-time.After(time.Millisecond):
t.Errorf("expected to close Connection.NotifyClose chan after Connection.Close")
}

select {
case <-ch.NotifyFlow(make(chan bool)):
case <-time.After(time.Millisecond):
Expand Down
6 changes: 6 additions & 0 deletions connection.go
Expand Up @@ -368,8 +368,14 @@ so that it will be necessary to consume the Channel from the caller in order to

To reconnect after a transport or protocol error, register a listener here and
re-run your setup process.

The chan provided must be a buffered channel of size 1.
*/
func (c *Connection) NotifyClose(receiver chan *Error) chan *Error {
if cap(receiver) != 1 {
panic("channel.NotifyClose expectes cap=1 buffered channel")
}

c.m.Lock()
defer c.m.Unlock()

Expand Down
33 changes: 2 additions & 31 deletions doc.go
Expand Up @@ -110,40 +110,11 @@ In order to be notified when a connection or channel gets closed, both
structures offer the possibility to register channels using
[Channel.NotifyClose] and [Connection.NotifyClose] functions:

notifyConnCloseCh := conn.NotifyClose(make(chan *amqp.Error))
notifyConnCloseCh := conn.NotifyClose(make(chan *amqp.Error, 1))

No errors will be sent in case of a graceful connection close. In case of a
non-graceful closure due to e.g. network issue, or forced connection closure
from the Management UI, the error will be notified synchronously by the library.

The error is sent synchronously to the channel, so that the flow will wait until
the receiver consumes from the channel. To avoid deadlocks in the library, it is
necessary to consume from the channels. This could be done inside a
different goroutine with a select listening on the two channels inside a for
loop like:

go func() {
for notifyConnClose != nil || notifyChanClose != nil {
select {
case err, ok := <-notifyConnClose:
if !ok {
notifyConnClose = nil
} else {
fmt.Printf("connection closed, error %s", err)
}
case err, ok := <-notifyChanClose:
if !ok {
notifyChanClose = nil
} else {
fmt.Printf("channel closed, error %s", err)
}
}
}
}()

Another approach is to use buffered channels:

notifyConnCloseCh := conn.NotifyClose(make(chan *amqp.Error, 1))
from the Management UI, the error will be notified by the library.

The library sends to notification channels just once. After sending a notification
to all channels, the library closes all registered notification channels. After
Expand Down
6 changes: 3 additions & 3 deletions integration_test.go
Expand Up @@ -1644,7 +1644,7 @@ func TestChannelExceptionWithCloseIssue43(t *testing.T) {
t.Cleanup(func() { conn.Close() })

go func() {
for err := range conn.NotifyClose(make(chan *Error)) {
for err := range conn.NotifyClose(make(chan *Error, 1)) {
t.Log(err.Error())
}
}()
Expand All @@ -1655,7 +1655,7 @@ func TestChannelExceptionWithCloseIssue43(t *testing.T) {
}

go func() {
for err := range c1.NotifyClose(make(chan *Error)) {
for err := range c1.NotifyClose(make(chan *Error, 1)) {
t.Log("Channel1 Close: " + err.Error())
}
}()
Expand All @@ -1666,7 +1666,7 @@ func TestChannelExceptionWithCloseIssue43(t *testing.T) {
}

go func() {
for err := range c2.NotifyClose(make(chan *Error)) {
for err := range c2.NotifyClose(make(chan *Error, 1)) {
t.Log("Channel2 Close: " + err.Error())
}
}()
Expand Down