Polishing

Author: acogoluegnes

src/main/java/reactor/rabbitmq/SendOptions.java

@@ -37,15 +37,20 @@ public class SendOptions {
     /**
      * The maximum number of in-flight records that are fetched
      * from the outbound record publisher while publisher confirms are pending.
+     * <p>
+     * The number of in-flight records is not limited by default.
      *
-     * @since 1.1.1
+     * @since 1.2.0
      */
     private Integer maxInFlight;
 
     /**
-     * The scheduler used for publishing send results.
+     * The scheduler used for publishing publisher confirms when in-flight records are limited.
+     * <p>
+     * The default is {@link Schedulers#immediate()}, so the caller's thread.
      *
-     * @since 1.1.1
+     * @see #maxInFlight
+     * @since 1.2.0
      */
     private Scheduler scheduler = Schedulers.immediate();
 
@@ -68,7 +73,7 @@ public class SendOptions {
      * from the outbound record publisher while publisher confirms are pending.
      *
      * @return maximum number of in-flight records
-     * @since 1.1.1
+     * @since 1.2.0
      */
     public Integer getMaxInFlight() {
         return maxInFlight;
@@ -77,10 +82,12 @@ public Integer getMaxInFlight() {
     /**
      * Set the maximum number of in-flight records that are fetched
      * from the outbound record publisher while publisher confirms are pending.
+     * <p>
+     * The number of in-flight records is not limited by default.
      *
      * @param maxInFlight
      * @return this {@link SendOptions} instance
-     * @since 1.1.1
+     * @since 1.2.0
      */
     public SendOptions maxInFlight(int maxInFlight) {
         this.maxInFlight = maxInFlight;
@@ -95,7 +102,7 @@ public SendOptions maxInFlight(int maxInFlight) {
      * @param maxInFlight
      * @param scheduler
      * @return this {@link SendOptions} instance
-     * @since 1.1.1
+     * @since 1.2.0
      */
     public SendOptions maxInFlight(int maxInFlight, Scheduler scheduler) {
         this.maxInFlight = maxInFlight;
@@ -107,7 +114,7 @@ public SendOptions maxInFlight(int maxInFlight, Scheduler scheduler) {
      * The scheduler used for publishing send results.
      *
      * @return scheduler used for publishing send results
-     * @since 1.1.1
+     * @since 1.2.0
      */
     public Scheduler getScheduler() {
         return scheduler;

src/test/java/reactor/rabbitmq/RabbitFluxTests.java

@@ -726,17 +726,17 @@ public void publishConfirmsBackpressure() throws Exception {
 
         sender = createSender();
         sender.sendWithPublishConfirms(msgFlux).subscribe(new BaseSubscriber<OutboundMessageResult>() {
-                @Override
-                protected void hookOnSubscribe(Subscription subscription) {
-                    subscription.request(subscriberRequest);
-                }
+            @Override
+            protected void hookOnSubscribe(Subscription subscription) {
+                subscription.request(subscriberRequest);
+            }
 
-                @Override
-                protected void hookOnNext(OutboundMessageResult outboundMessageResult) {
-                    if (outboundMessageResult.getOutboundMessage() != null) {
-                        confirmedLatch.countDown();
-                    }
+            @Override
+            protected void hookOnNext(OutboundMessageResult outboundMessageResult) {
+                if (outboundMessageResult.getOutboundMessage() != null) {
+                    confirmedLatch.countDown();
                 }
+            }
         });
 
         assertTrue(consumedLatch.await(1, TimeUnit.SECONDS));
@@ -768,20 +768,21 @@ void publishConfirmsMaxInFlight() throws InterruptedException {
 
         Flux<OutboundMessage> msgFlux = Flux.range(0, nbMessages).map(i -> {
             int current = inflight.incrementAndGet();
-            if (current > maxInflight.get())
+            if (current > maxInflight.get()) {
                 maxInflight.set(current);
+            }
             return new OutboundMessage("", queue, "".getBytes());
         });
 
         sender = createSender();
         sender
-            .sendWithPublishConfirms(msgFlux, new SendOptions().maxInFlight(maxConcurrency))
-            .subscribe(outboundMessageResult -> {
-                inflight.decrementAndGet();
-                if (outboundMessageResult.isAck() && outboundMessageResult.getOutboundMessage() != null) {
-                    confirmedLatch.countDown();
-                }
-        });
+                .sendWithPublishConfirms(msgFlux, new SendOptions().maxInFlight(maxConcurrency))
+                .subscribe(outboundMessageResult -> {
+                    inflight.decrementAndGet();
+                    if (outboundMessageResult.isAck() && outboundMessageResult.getOutboundMessage() != null) {
+                        confirmedLatch.countDown();
+                    }
+                });
 
         assertTrue(confirmedLatch.await(1, TimeUnit.SECONDS));
         assertThat(maxInflight.get()).isLessThanOrEqualTo(maxConcurrency);
@@ -814,7 +815,7 @@ public void publishConfirmsErrorWhilePublishing() throws Exception {
         sender.sendWithPublishConfirms(msgFlux, new SendOptions().exceptionHandler((ctx, e) -> {
             throw new RabbitFluxException(e);
         }))   // Before change: (onNext -> onError -> onNext )
-              // After change (maxInFlight): (onNext -> onError)
+                // After change (maxInFlight): (onNext -> onError)
                 .subscribe(outboundMessageResult -> {
                             if (outboundMessageResult.getOutboundMessage() != null) {
                                 confirmLatch.countDown();