[ServiceBus] Cache UpdateDispositionWorkItem Mono (#22317)

Azure · Jun 22, 2021 · c0d060f · c0d060f
1 parent 23260e7
commit c0d060f
Showing 1 changed file with 2 additions and 2 deletions.
diff --git a/...rc/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorReceiver.java b/...rc/main/java/com/azure/messaging/servicebus/implementation/ServiceBusReactorReceiver.java
@@ -239,7 +239,7 @@ private Mono<Void> updateDispositionInternal(String lockToken, DeliveryState del
         }
 
         final UpdateDispositionWorkItem workItem = new UpdateDispositionWorkItem(lockToken, deliveryState, timeout);
-        final Mono<Void> result = Mono.create(sink -> {
+        final Mono<Void> result = Mono.<Void>create(sink -> {
             workItem.start(sink);
             try {
                 provider.getReactorDispatcher().invoke(() -> {
@@ -250,7 +250,7 @@ private Mono<Void> updateDispositionInternal(String lockToken, DeliveryState del
                 sink.error(new AmqpException(false, "updateDisposition failed while dispatching to Reactor.",
                     error, handler.getErrorContext(receiver)));
             }
-        });
+        }).cache();  // cache because closeAsync use `when` to subscribe this Mono again.
 
         workItem.setMono(result);