[ServiceBus] Cache UpdateDispositionWorkItem Mono

[YijunXieMS]

fixes #22299

ServiceBusReactorReceiver.updateDisposition does not return or throw error if try timeout is long (from about 50 seconds but no exact number, usually 10 seconds is safe) and network is down.
This is because when try timeout is long enough, ServiceBusReactorReceiver.updateDisposition() hasn't timed out while another thread closes the ServiceBusReactorReceiver. In the closeAsync method, the Mono object of work items of type UpdateDispositionWorkItem is subscribed again (using when). The dual subscription causes the first Mono's sink doesn't complete. Adding a .cache() to the Mono will ensure the sink object isn't created for a second time.

[conniey]

Do we need .then(Mono.empty())? If so, then use then()?

[YijunXieMS]

Good catch. Mono.empty() is not required here.

[conniey]

Why do we need it to be a MonoSink of Object? I thought you could cache to .<Void>cache()?

[YijunXieMS]

the .cache() call turns the sink into a MonoSink<Object> from MonoSink<Void>.

final Mono<Void> result = Mono.create(sink -> {
            workItem.start(sink);
            try {
                provider.getReactorDispatcher().invoke(() -> {
                    unsettled.disposition(deliveryState);
                    pendingUpdates.put(lockToken, workItem);
                });
            } catch (IOException error) {
                sink.error(new AmqpException(false, "updateDisposition failed while dispatching to Reactor.",
                    error, handler.getErrorContext(receiver)));
            }
        }).cache().then();

[srnagar]

You can cast the create operator to Void as shown below.

final Mono<Void> result = Mono.<Void>create(sink -> {
            workItem.start(sink);
            try {
                provider.getReactorDispatcher().invoke(() -> {
                    unsettled.disposition(deliveryState);
                    pendingUpdates.put(lockToken, workItem);
                });
            } catch (IOException error) {
                sink.error(new AmqpException(false, "updateDisposition failed while dispatching to Reactor.",
                    error, handler.getErrorContext(receiver)));
            }
        }).cache();