Add test for negative credits

Signed-off-by: Tim Terlegård <[email protected]>

Author: terletimehm

smallrye-reactive-messaging-amqp/src/main/java/io/smallrye/reactive/messaging/amqp/AmqpCreditBasedSender.java

@@ -168,7 +168,7 @@ private long setCreditsAndRequest(AmqpSender sender) {
             subscription.request(request);
             return credits;
         }
-        if (credits == 0L && subscription != Subscriptions.CANCELLED) {
+        if (credits <= 0L && subscription != Subscriptions.CANCELLED) {
             onNoMoreCredit(sender);
         }
         return 0L;
@@ -195,7 +195,7 @@ public void onNext(Message<?> message) {
                         if (tuple != null) { // No serialization issue
                             subscriber.onNext(tuple.getItem1());
                             long remainingCredits = tuple.getItem3();
-                            if (remainingCredits == 0) { // no more credit, request more
+                            if (remainingCredits <= 0) { // no more credit, request more
                                 onNoMoreCredit(tuple.getItem2());
                             } else { // keep the request one more message
                                 requestUpstream();
@@ -221,9 +221,9 @@ private void onNoMoreCredit(AmqpSender sender) {
                     return;
                 }
                 long c = setCreditsAndRequest(sender);
-                if (c <= 0L) { // still no credits, schedule a periodic retry
+                if (c == 0L) { // still no credits, schedule a periodic retry
                     holder.getVertx().setPeriodic(configuration.getCreditRetrievalPeriod(), id -> {
-                        if (setCreditsAndRequest(sender) > 0L || isCancelled()) {
+                        if (setCreditsAndRequest(sender) != 0L || isCancelled()) {
                             // Got our new credits or the application has been terminated,
                             // we cancel the periodic task.
                             holder.getVertx().cancelTimer(id);

smallrye-reactive-messaging-amqp/src/test/java/io/smallrye/reactive/messaging/amqp/AmqpCreditTest.java

@@ -2,6 +2,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -10,6 +11,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Flow;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -22,8 +24,10 @@
 import org.junit.jupiter.api.Timeout;
 
 import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
 import io.smallrye.reactive.messaging.test.common.config.MapBasedConfig;
 import io.vertx.core.Vertx;
+import io.vertx.proton.ProtonReceiver;
 
 public class AmqpCreditTest extends AmqpTestBase {
 
@@ -48,10 +52,36 @@ public void testCreditBasedFlowControl() throws Exception {
         CountDownLatch msgsReceived = new CountDownLatch(msgCount);
         List<Object> payloadsReceived = new ArrayList<>(msgCount);
 
-        server = setupMockServer(msgCount, msgsReceived, payloadsReceived, executionHolder.vertx().getDelegate());
+        server = setupMockServer(msgCount, msgsReceived, payloadsReceived, executionHolder.vertx().getDelegate(),
+                msgCount / 10, r -> {
+                });
+
+        Flow.Subscriber<? extends Message<?>> sink = createProviderAndSink(UUID.randomUUID().toString(),
+                server.actualPort());
+        //noinspection unchecked
+        Multi.createFrom().range(0, msgCount)
+                .map(Message::of)
+                .subscribe((Flow.Subscriber<? super Message<Integer>>) sink);
+
+        assertThat(msgsReceived.await(20, TimeUnit.SECONDS))
+                .withFailMessage("Sent %s msgs but %s remain outstanding", msgCount, msgsReceived.getCount()).isTrue();
+        List<Integer> expectedPayloads = IntStream.range(0, msgCount).boxed().collect(Collectors.toList());
+        assertThat(payloadsReceived).containsAll(expectedPayloads);
+    }
+
+    @Test
+    @Timeout(30)
+    public void testNegativeCredit() throws Exception {
+        int msgCount = 10;
+        CountDownLatch msgsReceived = new CountDownLatch(msgCount);
+        List<Object> payloadsReceived = new ArrayList<>(msgCount);
+
+        server = setupMockServer(msgCount, msgsReceived, payloadsReceived, executionHolder.vertx().getDelegate(), -2,
+                this::updateCreditsEveryTwoSeconds);
 
         Flow.Subscriber<? extends Message<?>> sink = createProviderAndSink(UUID.randomUUID().toString(),
                 server.actualPort());
+
         //noinspection unchecked
         Multi.createFrom().range(0, msgCount)
                 .map(Message::of)
@@ -78,7 +108,18 @@ private Flow.Subscriber<? extends Message<?>> createProviderAndSink(String topic
         return provider.getSubscriber(new MapBasedConfig(config));
     }
 
-    private MockServer setupMockServer(int msgCount, CountDownLatch latch, List<Object> payloads, Vertx vertx)
+    private void updateCreditsEveryTwoSeconds(ProtonReceiver serverReceiver) {
+        Multi.createFrom().items(2, 4, 6)
+                .onItem().transformToUniAndMerge(seconds -> {
+                    Duration delay = Duration.ofSeconds(seconds);
+                    return Uni.createFrom().item(seconds)
+                            .onItem().delayIt().by(delay);
+                })
+                .subscribe().with(item -> serverReceiver.flow(1));
+    }
+
+    private MockServer setupMockServer(int msgCount, CountDownLatch latch, List<Object> payloads, Vertx vertx,
+            int initialCredits, Consumer<ProtonReceiver> receiverFunction)
             throws Exception {
         assertThat(msgCount % 10 == 0).isTrue();
         int creditBatch = msgCount / 10;
@@ -114,9 +155,11 @@ private MockServer setupMockServer(int msgCount, CountDownLatch latch, List<Obje
                     }
                 });
 
+                receiverFunction.accept(serverReceiver);
+
                 serverReceiver.open();
 
-                serverReceiver.flow(creditBatch);
+                serverReceiver.flow(initialCredits);
             });
         });
     }