Add Done and Err methods to Conn (#348)

* Add Done and Err methods to Conn

Signals when a connection has been closed and any associated error.

* fix up changelog (bad merge)

* add some more comments and an example

* refine comments

* fix type-o in changelog

Author: jhendrixMSFT

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Release History
 
+## 1.3.0 (unreleased)
+
+### Features Added
+
+* Added methods `Done` and `Err` to `Conn`
+  * `Done` returns a channel that's closed when `Conn` has closed.
+  * `Err` explains why `Conn` was closed.
+
 ## 1.2.0 (2024-09-30)
 
 ### Features Added
@@ -9,7 +17,7 @@
 
 ### Bugs Fixed
 
-* Fixed a rare race in `Conn.start` that could cause goroutines to be leaked if the provided context was canceld/expired.
+* Fixed a rare race in `Conn.start` that could cause goroutines to be leaked if the provided context was canceled/expired.
 
 ### Other Changes
 

conn.go

@@ -406,6 +406,13 @@ func (c *Conn) startImpl(ctx context.Context) error {
 }
 
 // Close closes the connection.
+//
+// Returns nil if there were no errors during shutdown,
+// or a *ConnError. This error is not actionable and is
+// purely for diagnostic purposes.
+//
+// The error returned by subsequent calls to Close is
+// idempotent, so the same value will always be returned.
 func (c *Conn) Close() error {
 	c.close()
 
@@ -415,15 +422,32 @@ func (c *Conn) Close() error {
 	<-c.txDone
 	<-c.rxDone
 
-	var connErr *ConnError
-	if errors.As(c.doneErr, &connErr) && connErr.RemoteErr == nil && connErr.inner == nil {
-		// an empty ConnectionError means the connection was closed by the caller
+	return c.closedErr()
+}
+
+// Done returns a channel that's closed when Conn is closed.
+func (c *Conn) Done() <-chan struct{} {
+	return c.done
+}
+
+// If Done is not yet closed, Err returns nil.
+// If Done is closed, Err returns nil or a *ConnError explaining why.
+// A nil error indicates that [Close] was called and there
+// were no errors during shutdown.
+//
+// A *ConnError indicates one of three things
+//   - there was an error during shutdown from a client-side call to [Close]. the
+//     error is not actionable and is purely for diagnostic purposes.
+//   - a fatal error was encountered that caused [Conn] to close
+//   - the peer closed the connection. [ConnError.RemoteErr] MAY contain an error
+//     from the peer indicating why it closed the connection
+func (c *Conn) Err() error {
+	select {
+	case <-c.done:
+		return c.closedErr()
+	default:
 		return nil
 	}
-
-	// there was an error during shut-down or connReader/connWriter
-	// experienced a terminal error
-	return c.doneErr
 }
 
 // close is called once, either from Close() or when connReader/connWriter exits
@@ -471,6 +495,20 @@ func (c *Conn) closeDuringStart() {
 	})
 }
 
+// returns the error indicating why Conn has closed
+// NOTE: only call this AFTER Conn.done has been closed!
+func (c *Conn) closedErr() error {
+	// an empty ConnError means the connection was closed by the caller
+	var connErr *ConnError
+	if errors.As(c.doneErr, &connErr) && connErr.RemoteErr == nil && connErr.inner == nil {
+		return nil
+	}
+
+	// there was an error during shut-down or connReader/connWriter
+	// experienced a terminal error
+	return c.doneErr
+}
+
 // NewSession starts a new session on the connection.
 //   - ctx controls waiting for the peer to acknowledge the session
 //   - opts contains optional values, pass nil to accept the defaults

conn_test.go

@@ -302,6 +302,13 @@ func TestClose(t *testing.T) {
 	cancel()
 	require.Nil(t, conn.Properties())
 	require.NoError(t, conn.Close())
+	select {
+	case <-conn.Done():
+		require.NoError(t, conn.Err())
+	default:
+		t.Fatal("expected conn.Done() to be signaled")
+	}
+
 	// with Close error
 	netConn = fake.NewNetConn(senderFrameHandlerNoUnhandled(0, SenderSettleModeUnsettled), fake.NetConnOptions{})
 	conn, err = newConn(netConn, nil)
@@ -315,6 +322,50 @@ func TestClose(t *testing.T) {
 	// wait a bit for connReader to read from the mock
 	time.Sleep(100 * time.Millisecond)
 	require.Error(t, conn.Close())
+	select {
+	case <-conn.Done():
+		require.Error(t, conn.Err())
+	default:
+		t.Fatal("expected conn.Done() to be signaled")
+	}
+}
+
+func TestCloseAsync(t *testing.T) {
+	netConn := fake.NewNetConn(senderFrameHandlerNoUnhandled(0, SenderSettleModeUnsettled), fake.NetConnOptions{})
+	conn, err := newConn(netConn, nil)
+	require.NoError(t, err)
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	require.NoError(t, conn.start(ctx))
+	cancel()
+	go func() {
+		require.NoError(t, conn.Close())
+	}()
+	select {
+	case <-conn.Done():
+		require.NoError(t, conn.Err())
+	case <-time.After(1 * time.Second):
+		t.Fatal("expected conn.Done() to be signaled")
+	}
+
+	// with Close error
+	netConn = fake.NewNetConn(senderFrameHandlerNoUnhandled(0, SenderSettleModeUnsettled), fake.NetConnOptions{})
+	conn, err = newConn(netConn, nil)
+	require.NoError(t, err)
+	ctx, cancel = context.WithTimeout(context.Background(), 1*time.Second)
+	require.NoError(t, conn.start(ctx))
+	cancel()
+	netConn.OnClose = func() error {
+		return errors.New("mock close failed")
+	}
+	go func() {
+		require.Error(t, conn.Close())
+	}()
+	select {
+	case <-conn.Done():
+		require.Error(t, conn.Err())
+	case <-time.After(1 * time.Second):
+		t.Fatal("expected conn.Done() to be signaled")
+	}
 }
 
 func TestServerSideClose(t *testing.T) {
@@ -344,6 +395,12 @@ func TestServerSideClose(t *testing.T) {
 	<-closeReceived
 	err = conn.Close()
 	require.NoError(t, err)
+	select {
+	case <-conn.Done():
+		require.NoError(t, conn.Err())
+	default:
+		t.Fatal("expected conn.Done() to be signaled")
+	}
 
 	// with error
 	closeReceived = make(chan struct{})
@@ -361,6 +418,16 @@ func TestServerSideClose(t *testing.T) {
 	var connErr *ConnError
 	require.ErrorAs(t, err, &connErr)
 	require.Equal(t, "*Error{Condition: Close, Description: mock server error, Info: map[]}", connErr.Error())
+	select {
+	case <-conn.Done():
+		connErr = nil
+		require.ErrorAs(t, conn.Err(), &connErr)
+		require.Equal(t, "*Error{Condition: Close, Description: mock server error, Info: map[]}", connErr.Error())
+	default:
+		t.Fatal("expected conn.Done() to be signaled")
+	}
+	require.ErrorAs(t, conn.Err(), &connErr)
+	require.Equal(t, "*Error{Condition: Close, Description: mock server error, Info: map[]}", connErr.Error())
 }
 
 func TestKeepAlives(t *testing.T) {
@@ -592,7 +659,9 @@ func TestClientClose(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, client)
 	require.NoError(t, client.Close())
+	require.NoError(t, client.Err())
 	require.NoError(t, client.Close())
+	require.NoError(t, client.Err())
 }
 
 func TestSessionOptions(t *testing.T) {

errors.go

@@ -83,7 +83,7 @@ type ConnError struct {
 	inner error
 }
 
-// Error implements the error interface for ConnectionError.
+// Error implements the error interface for ConnError.
 func (e *ConnError) Error() string {
 	if e.RemoteErr == nil && e.inner == nil {
 		return "amqp: connection closed"

example_test.go

@@ -218,3 +218,42 @@ func ExampleLinkError() {
 		log.Fatalf("unexpected error type %T", err)
 	}
 }
+
+func ExampleConn_Done() {
+	ctx := context.TODO()
+
+	// create connection
+	conn, err := amqp.Dial(ctx, "amqps://my-namespace.servicebus.windows.net", &amqp.ConnOptions{
+		SASLType: amqp.SASLTypePlain("access-key-name", "access-key"),
+	})
+	if err != nil {
+		log.Fatal("Dialing AMQP server:", err)
+	}
+
+	// when the channel returned by Done is closed, conn has been closed
+	<-conn.Done()
+
+	// Err indicates why the connection was closed. a nil error indicates
+	// a client-side call to Close and there were no errors during shutdown.
+	closedErr := conn.Err()
+
+	// when Err returns a non-nil error, it means that either a client-side
+	// call to Close encountered an error during shutdown, a fatal error was
+	// encountered that caused the connection to close, or that the peer
+	// closed the connection.
+	if closedErr != nil {
+		// the error returned by Err is always a *ConnError
+		var connErr *amqp.ConnError
+		errors.As(closedErr, &connErr)
+
+		if connErr.RemoteErr != nil {
+			// the peer closed the connection and provided an error explaining why.
+			// note that the peer MAY send an error when closing the connection but
+			// is not required to.
+		} else {
+			// the connection encountered a fatal error or there was
+			// an error during client-side shutdown. this is for
+			// diagnostics, the connection has been closed.
+		}
+	}
+}