Added test for subscriptions - checking publisher and how many is got asked for

bbakerman · bbakerman · commit bd4a0982ef18 · 2025-04-03T12:49:26.000+11:00
diff --git a/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy b/src/test/groovy/graphql/execution/SubscriptionExecutionStrategyTest.groovy
@@ -712,7 +712,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
 
     def "we can cancel the operation and the upstream publisher is told"() {
         List<Runnable> promises = []
-        Publisher<Object> publisher = new RxJavaMessagePublisher(10)
+        RxJavaMessagePublisher publisher = new RxJavaMessagePublisher(10)
 
         DataFetcher newMessageDF = { env -> return publisher }
         DataFetcher senderDF = dfThatDoesNotComplete("sender", promises)
@@ -750,6 +750,7 @@ class SubscriptionExecutionStrategyTest extends Specification {
         messages.size() == 1
         def error = messages[0].errors[0]
         assert error.message == "Execution has been asked to be cancelled"
+        publisher.counter == 2
     }
 
     private static DataFetcher<?> dfThatDoesNotComplete(String propertyName, List<Runnable> promises) {
diff --git a/src/test/groovy/graphql/execution/pubsub/RxJavaMessagePublisher.java b/src/test/groovy/graphql/execution/pubsub/RxJavaMessagePublisher.java
@@ -4,17 +4,27 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 
+import java.util.concurrent.atomic.AtomicInteger;
+
 /**
  * This example publisher will create count "messages" and then terminate. Its
  * uses tRxJava Flowable as a backing publisher
  */
 public class RxJavaMessagePublisher implements Publisher<Message> {
 
     private final Flowable<Message> flowable;
+    private final AtomicInteger counter = new AtomicInteger();
 
     public RxJavaMessagePublisher(final int count) {
         flowable = Flowable.range(0, count)
-                .map(at -> examineMessage(new Message("sender" + at, "text" + at), at));
+                .map(at -> {
+                    counter.incrementAndGet();
+                    return examineMessage(new Message("sender" + at, "text" + at), at);
+                });
+    }
+
+    public int getCounter() {
+        return counter.get();
     }
 
     @Override
