test: add GCP pubsub conformance tests for both push delivery and fanout scenarios (#243)

Author: roseyang62

pubsub/pubsub-aws/src/test/java/com/salesforce/multicloudj/pubsub/aws/AwsPubsubIT.java

@@ -22,12 +22,13 @@
 import java.util.concurrent.ThreadLocalRandom;
 
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 
 public class AwsPubsubIT extends AbstractPubsubIT {
 
     private static final String SQS_ENDPOINT = "https://sqs.us-west-2.amazonaws.com";
-    private static final String ACCOUNT_ID = "654654370895";
     private static final String BASE_QUEUE_NAME = "test-queue";
 
     private HarnessImpl harnessImpl;
@@ -202,4 +203,12 @@ public void close() throws Exception {
             }
         }
     }
+
+    // TODO: Implement testSendReceiveTwo() when WireMock adds support for multiple upstream endpoints.
+    // The behavior has been validated through manual testing for now.
+    @Override
+    @Test
+    @Disabled()
+    public void testSendReceiveTwo() throws Exception {
+    }
 }

pubsub/pubsub-client/src/test/java/com/salesforce/multicloudj/pubsub/client/AbstractPubsubIT.java

@@ -18,6 +18,8 @@
 import org.junit.jupiter.api.Timeout;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
@@ -29,7 +31,19 @@ public interface Harness extends AutoCloseable {
 
         AbstractTopic createTopicDriver();
 
-        AbstractSubscription createSubscriptionDriver();
+        default AbstractSubscription createSubscriptionDriver() {
+            return createSubscriptionDriverWithIndex(0);
+        }
+
+        /**
+         * Create a subscription driver with an index suffix.
+         * This allows creating multiple subscriptions to the same topic.
+         */
+        default AbstractSubscription createSubscriptionDriverWithIndex(int index) {
+            // Default implementation: just create a subscription driver
+            // Providers that need index-based naming should override this method
+            return createSubscriptionDriver();
+        }
 
         String getPubsubEndpoint();
 
@@ -375,4 +389,124 @@ public void testMultipleSendReceiveWithoutBatch() throws Exception {
             }
         }
     }
+
+    /**
+     * Receive from two subscriptions to the same topic.
+     * Verify both get all the messages.
+     */
+    @Test
+    @Timeout(120) // Integration test with multiple subscriptions - allow time for message delivery
+    public void testSendReceiveTwo() throws Exception {
+        // Create two subscriptions to the same topic
+        AbstractSubscription subscription1 = harness.createSubscriptionDriverWithIndex(1);
+        AbstractSubscription subscription2 = harness.createSubscriptionDriverWithIndex(2);
+
+        try (AbstractTopic topic = harness.createTopicDriver();
+             AbstractSubscription sub1 = subscription1;
+             AbstractSubscription sub2 = subscription2) {
+
+            // Send 3 messages to the topic
+            List<Message> messagesToSend = List.of(
+                    Message.builder().withBody("fanout-msg1".getBytes()).withMetadata(Map.of("id", "1")).build(),
+                    Message.builder().withBody("fanout-msg2".getBytes()).withMetadata(Map.of("id", "2")).build(),
+                    Message.builder().withBody("fanout-msg3".getBytes()).withMetadata(Map.of("id", "3")).build()
+            );
+
+            for (Message message : messagesToSend) {
+                topic.send(message);
+            }
+
+            TimeUnit.MILLISECONDS.sleep(500);
+
+            // Receive messages from both subscriptions
+            List<Message> received1 = receiveMessages(sub1, messagesToSend.size());
+            List<Message> received2 = receiveMessages(sub2, messagesToSend.size());
+
+            // Verify both subscriptions received all messages
+            Assertions.assertEquals(messagesToSend.size(), received1.size(),
+                    "Subscription 1 should receive all " + messagesToSend.size() + " messages. Got: " + received1.size());
+            Assertions.assertEquals(messagesToSend.size(), received2.size(),
+                    "Subscription 2 should receive all " + messagesToSend.size() + " messages. Got: " + received2.size());
+
+            // Verify messages match for both subscriptions
+            verifyMessages(received1, messagesToSend, "Subscription 1");
+            verifyMessages(received2, messagesToSend, "Subscription 2");
+
+            // Ack all messages from both subscriptions
+            ackMessages(sub1, received1);
+            ackMessages(sub2, received2);
+        }
+    }
+
+    /**
+     * Helper function: Receives messages from a subscription until the expected count is reached.
+     */
+    private List<Message> receiveMessages(AbstractSubscription subscription, int expectedCount) throws InterruptedException {
+        boolean isRecording = System.getProperty("record") != null;
+        long timeoutSeconds = isRecording ? 120 : 60;
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds);
+
+        List<Message> received = new ArrayList<>();
+        while (received.size() < expectedCount && System.nanoTime() < deadline) {
+            try {
+                Message r = subscription.receive();
+                if (r != null && r.getAckID() != null) {
+                    received.add(r);
+                } else {
+                    TimeUnit.MILLISECONDS.sleep(100);
+                }
+            } catch (Exception e) {
+                TimeUnit.MILLISECONDS.sleep(100);
+            }
+        }
+        return received;
+    }
+
+    /**
+     * Helper function: Verifies that received messages match the expected messages.
+     */
+    private void verifyMessages(List<Message> received, List<Message> expected, String subscriptionName) {
+        if (received.size() != expected.size()) {
+            Assertions.fail(String.format("%s: got %d messages, expected %d", 
+                    subscriptionName, received.size(), expected.size()));
+        }
+        Map<String, Message> gotByBody = new HashMap<>();
+        for (Message msg : received) {
+            gotByBody.put(new String(msg.getBody()), msg);
+        }
+        for (Message exp : expected) {
+            String body = new String(exp.getBody());
+            Message got = gotByBody.get(body);
+            if (got == null) {
+                Assertions.fail(subscriptionName + ": missing message: " + body);
+            }
+            if (!Arrays.equals(exp.getBody(), got.getBody())) {
+                Assertions.fail(subscriptionName + ": body mismatch for " + body);
+            }
+            if (exp.getMetadata() != null) {
+                for (Map.Entry<String, String> entry : exp.getMetadata().entrySet()) {
+                    String expValue = entry.getValue();
+                    String gotValue = got.getMetadata() != null ? got.getMetadata().get(entry.getKey()) : null;
+                    if (!expValue.equals(gotValue)) {
+                        Assertions.fail(String.format("%s: metadata[%s] mismatch for %s: expected %s, got %s",
+                                subscriptionName, entry.getKey(), body, expValue, gotValue));
+                    }
+                }
+            }
+        }
+    }
+
+
+    /**
+     * Helper function: Acknowledges all messages in the given list.
+     */
+    private void ackMessages(AbstractSubscription subscription, List<Message> messages) {
+        if (!messages.isEmpty()) {
+            List<AckID> ackIDs = new ArrayList<>();
+            for (Message msg : messages) {
+                ackIDs.add(msg.getAckID());
+            }
+            subscription.sendAcks(ackIDs).join();
+        }
+    }
 }