test(pubsub): cleanup ordered keys json integration test (#8575)

hongalex · web-flow · commit 53f17c4380cd · 2023-09-21T16:33:09.000-07:00
* test(pubsub): remove checking of publish result to avoid creating too many goroutines

* add check for receive to exit before deleting subscription
diff --git a/pubsub/integration_test.go b/pubsub/integration_test.go
@@ -1263,23 +1263,18 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
 		key := parts[0]
 		msg := parts[1]
 		publishData = append(publishData, testutil2.OrderedKeyMsg{Key: key, Data: msg})
-		r := topic.Publish(ctx, &Message{
+		topic.Publish(ctx, &Message{
 			Data:        []byte(msg),
 			OrderingKey: key,
 		})
-		go func() {
-			_, err := r.Get(ctx)
-			if err != nil {
-				// Can't fail inside goroutine, so just log the error.
-				t.Logf("publish error for message(%s): %v", msg, err)
-			}
-		}()
 		wg.Add(1)
 	}
 	if err := scanner.Err(); err != nil {
 		t.Fatal(err)
 	}
 
+	receiveDone := make(chan struct{})
+	ctx, cancel := context.WithCancel(ctx)
 	go func() {
 		if err := sub.Receive(ctx, func(ctx context.Context, msg *Message) {
 			mu.Lock()
@@ -1298,6 +1293,7 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
 				t.Error(err)
 			}
 		}
+		close(receiveDone)
 	}()
 
 	done := make(chan struct{})
@@ -1308,6 +1304,7 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
 
 	select {
 	case <-done:
+		cancel()
 	case <-time.After(5 * time.Minute):
 		t.Fatal("timed out after 5m waiting for all messages to be received")
 	}
@@ -1317,6 +1314,13 @@ func TestIntegration_OrderedKeys_JSON(t *testing.T) {
 	if err := testutil2.VerifyKeyOrdering(publishData, receiveData); err != nil {
 		t.Fatalf("VerifyKeyOrdering error: %v", err)
 	}
+
+	select {
+	case <-receiveDone:
+	case <-time.After(5 * time.Minute):
+		t.Fatal("timed out after 5m waiting for receive to exit")
+	}
+
 }
 
 func TestIntegration_OrderedKeys_ResumePublish(t *testing.T) {
