fix(pubsub): suppress connection is closing error (#2951)

pubsub/pubsub.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"os"
 	"runtime"
+	"strings"
 	"time"
 
 	"cloud.google.com/go/internal/version"
@@ -101,10 +102,17 @@ func (c *Client) Close() error {
 	pubErr := c.pubc.Close()
 	subErr := c.subc.Close()
 	if pubErr != nil {
-		return fmt.Errorf("pubsub subscriber closing error: %v", pubErr)
+		return fmt.Errorf("pubsub publisher closing error: %v", pubErr)
 	}
 	if subErr != nil {
-		return fmt.Errorf("pubsub publisher closing error: %v", subErr)
+		// Suppress client connection closing errors. This will only happen
+		// when using the client in conjunction with the Pub/Sub emulator
+		// or fake (pstest). Closing both clients separately will never
+		// return this error against the live Pub/Sub service.
+		if strings.Contains(subErr.Error(), "the client connection is closing") {
+			return nil
+		}
+		return fmt.Errorf("pubsub subscriber closing error: %v", subErr)
 	}
 	return nil
 }

pubsub/streaming_pull_test.go

@@ -315,12 +315,9 @@ func TestStreamingPull_ClosedClient(t *testing.T) {
 	// wait for receives to happen
 	time.Sleep(100 * time.Millisecond)
 
-	// Intentionally don't check the returned err here. In the fake,
-	// closing either the publisher/subscriber client will cause the
-	// server to clean up resources, which is different than in the
-	// live service. With the fake, client.Close() will return a
-	// "the client connection is closing" error with the second Close.
-	client.Close()
+	if err := client.Close(); err != nil {
+		t.Fatalf("Got error while closing client: %v", err)
+	}
 
 	// wait for things to close
 	time.Sleep(100 * time.Millisecond)