Allow and handle negative credits, fixes invalid request number error (#3213)

Author: terletimehm

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

@@ -161,14 +161,14 @@ public void onSubscribe(Subscription subscription) {
     private long setCreditsAndRequest(AmqpSender sender) {
         long credits = sender.remainingCredits();
         Subscription subscription = upstream.get();
-        if (credits != 0L && subscription != Subscriptions.CANCELLED) {
+        if (credits > 0L && subscription != Subscriptions.CANCELLED) {
             // Request upfront the sender remaining credits or the max inflights
             long request = maxInflights > 0 ? Math.min(credits, maxInflights) : credits;
             log.retrievedCreditsForChannel(configuration.getChannel(), credits);
             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();

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,8 +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)
-            throws Exception {
+    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 +154,11 @@ private MockServer setupMockServer(int msgCount, CountDownLatch latch, List<Obje
                     }
                 });
 
+                receiverFunction.accept(serverReceiver);
+
                 serverReceiver.open();
 
-                serverReceiver.flow(creditBatch);
+                serverReceiver.flow(initialCredits);
             });
         });
     }