diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java index 3623431604..af2edff793 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java @@ -29,7 +29,6 @@ import org.springframework.util.Assert; import reactor.core.Disposable; import reactor.core.Disposables; import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxProcessor; import reactor.core.publisher.Sinks; import reactor.core.scheduler.Schedulers; @@ -46,9 +45,7 @@ import reactor.core.scheduler.Schedulers; public class FluxMessageChannel extends AbstractMessageChannel implements Publisher>, ReactiveStreamsSubscribableChannel { - private final Sinks.Many> sink; - - private final FluxProcessor, Message> processor; + private final Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(1, false); private final Sinks.Many subscribedSignal = Sinks.many().replay().limit(1); @@ -56,14 +53,9 @@ public class FluxMessageChannel extends AbstractMessageChannel private volatile boolean active = true; - public FluxMessageChannel() { - this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false); - this.processor = FluxProcessor.fromSink(this.sink); - } - @Override protected boolean doSend(Message message, long timeout) { - Assert.state(this.active && this.processor.hasDownstreams(), + Assert.state(this.active && this.sink.currentSubscriberCount() > 0, () -> "The [" + this + "] doesn't have subscribers to accept messages"); long remainingTime = 0; if (timeout > 0) { @@ -101,11 +93,12 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void subscribe(Subscriber> subscriber) { - this.processor - .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams())) + this.sink.asFlux() + .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0)) .share() .subscribe(subscriber); - this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams()); + + this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0); } @Override @@ -131,9 +124,9 @@ public class FluxMessageChannel extends AbstractMessageChannel @Override public void destroy() { this.active = false; - this.subscribedSignal.tryEmitNext(false); this.upstreamSubscriptions.dispose(); - this.processor.onComplete(); + this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); + this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST); super.destroy(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index be00b75523..dc5f9d0066 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -67,7 +67,6 @@ import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoProcessor; import reactor.core.publisher.Sinks; /** @@ -899,7 +898,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint @Override public void subscribeTo(Publisher> publisher) { - publisher.subscribe(MonoProcessor.fromSink(this.replyMono)); + Mono.from(publisher) + .subscribe( + (value) -> this.replyMono.emitValue(value, Sinks.EmitFailureHandler.FAIL_FAST), + this.replyMono::tryEmitError, this.replyMono::tryEmitEmpty); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java index f996930562..2a5d6eb415 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java @@ -51,7 +51,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import reactor.core.Disposable; import reactor.core.publisher.Flux; -import reactor.core.publisher.FluxProcessor; /** * @author Artem Bilan @@ -141,7 +140,7 @@ public class FluxMessageChannelTests { flowRegistration.destroy(); - assertThat(TestUtils.getPropertyValue(flux, "processor", FluxProcessor.class).isTerminated()).isTrue(); + assertThat(TestUtils.getPropertyValue(flux, "sink.sink.done", Boolean.class)).isTrue(); } @Configuration diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java index c730068850..f120e8427b 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java @@ -17,24 +17,35 @@ package org.springframework.integration.rsocket; import java.lang.reflect.Method; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicReference; import org.springframework.context.ApplicationContext; import org.springframework.core.MethodParameter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.codec.Encoder; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.handler.CompositeMessageCondition; import org.springframework.messaging.handler.DestinationPatternsMessageCondition; import org.springframework.messaging.handler.invocation.reactive.HandlerMethodArgumentResolver; +import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler; import org.springframework.messaging.handler.invocation.reactive.SyncHandlerMethodArgumentResolver; import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition; import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; +import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler; +import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; +import io.rsocket.Payload; import io.rsocket.frame.FrameType; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * The {@link RSocketMessageHandler} extension for Spring Integration needs. @@ -109,6 +120,23 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler { } } + @Override + @SuppressWarnings("unchecked") + protected List initReturnValueHandlers() { + HandlerMethodReturnValueHandler integrationRSocketPayloadReturnValueHandler = + new IntegrationRSocketPayloadReturnValueHandler((List>) getEncoders(), + getReactiveAdapterRegistry()); + if (this.messageMappingCompatible) { + List handlers = new ArrayList<>(); + handlers.add(integrationRSocketPayloadReturnValueHandler); + handlers.addAll(getReturnValueHandlerConfigurer().getCustomHandlers()); + return handlers; + } + else { + return Collections.singletonList(integrationRSocketPayloadReturnValueHandler); + } + } + protected static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver { @Override @@ -123,4 +151,35 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler { } + protected static final class IntegrationRSocketPayloadReturnValueHandler extends RSocketPayloadReturnValueHandler { + + protected IntegrationRSocketPayloadReturnValueHandler(List> encoders, + ReactiveAdapterRegistry registry) { + + super(encoders, registry); + } + + @Override public Mono handleReturnValue(@Nullable Object returnValue, MethodParameter returnType, + Message message) { + + AtomicReference> responseReference = getResponseReference(message); + + if (returnValue == null && responseReference != null) { + return super.handleReturnValue(responseReference.get(), returnType, message); + } + else { + return super.handleReturnValue(returnValue, returnType, message); + } + } + + @Nullable + @SuppressWarnings("unchecked") + private static AtomicReference> getResponseReference(Message message) { + Object headerValue = message.getHeaders().get(RESPONSE_HEADER); + Assert.state(headerValue == null || headerValue instanceof AtomicReference, "Expected AtomicReference"); + return (AtomicReference>) headerValue; + } + + } + } diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java deleted file mode 100644 index 23cd0fa23c..0000000000 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java +++ /dev/null @@ -1,467 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.rsocket.inbound; - -import java.util.function.Function; - -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; - -import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.DataBufferUtils; -import org.springframework.lang.Nullable; -import org.springframework.util.Assert; - -import reactor.core.CoreSubscriber; -import reactor.core.Scannable; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.Operators; -import reactor.util.context.Context; - -/** - * ---------------------- - *

NOTE: This class was copied from - * {@code org.springframework.http.server.reactive.ChannelSendOperator} - * & {@code org.springframework.messaging.handler.invocation.reactive.ChannelSendOperator} - * and is identical to them. It's used for the same purpose, i.e. the ability to switch to - * alternate handling via annotated exception handler methods if the output - * publisher starts with an error. - *

----------------------
- * - *

Given a write function that accepts a source {@code Publisher} to write - * with and returns {@code Publisher} for the result, this operator helps - * to defer the invocation of the write function, until we know if the source - * publisher will begin publishing without an error. If the first emission is - * an error, the write function is bypassed, and the error is sent directly - * through the result publisher. Otherwise the write function is invoked. - * - * @author Rossen Stoyanchev - * @author Stephane Maldini - * @author Artem Bilan - * - * @since 5.2 - * - * @param the type of element signaled - */ -class ChannelSendOperator extends Mono implements Scannable { - - private final Function, Publisher> writeFunction; - - private final Flux source; - - - ChannelSendOperator(Publisher source, Function, Publisher> writeFunction) { - this.source = Flux.from(source); - this.writeFunction = writeFunction; - } - - - @Override - @Nullable - @SuppressWarnings("rawtypes") - public Object scanUnsafe(Attr key) { - if (key == Attr.PREFETCH) { - return Integer.MAX_VALUE; - } - if (key == Attr.PARENT) { - return this.source; - } - return null; - } - - @Override - public void subscribe(CoreSubscriber actual) { - this.source.subscribe(new WriteBarrier(actual)); - } - - - private enum State { - - /** No emissions from the upstream source yet. */ - NEW, - - /** - * At least one signal of any kind has been received; we're ready to - * call the write function and proceed with actual writing. - */ - FIRST_SIGNAL_RECEIVED, - - /** - * The write subscriber has subscribed and requested; we're going to - * emit the cached signals. - */ - EMITTING_CACHED_SIGNALS, - - /** - * The write subscriber has subscribed, and cached signals have been - * emitted to it; we're ready to switch to a simple pass-through mode - * for all remaining signals. - **/ - READY_TO_WRITE - - } - - - /** - * A barrier inserted between the write source and the write subscriber - * (i.e. the HTTP server adapter) that pre-fetches and waits for the first - * signal before deciding whether to hook in to the write subscriber. - * - *

Acts as: - *

    - *
  • Subscriber to the write source. - *
  • Subscription to the write subscriber. - *
  • Publisher to the write subscriber. - *
- * - *

Also uses {@link WriteCompletionBarrier} to communicate completion - * and detect cancel signals from the completion subscriber. - */ - private class WriteBarrier implements CoreSubscriber, Subscription, Publisher { - - /* Bridges signals to and from the completionSubscriber */ - private final WriteCompletionBarrier writeCompletionBarrier; - - /* Upstream write source subscription */ - @Nullable - private Subscription subscription; - - /** Cached data item before readyToWrite. */ - @Nullable - private T item; - - /** Cached error signal before readyToWrite. */ - @Nullable - private Throwable error; - - /** Cached onComplete signal before readyToWrite. */ - private boolean completed = false; - - /** Recursive demand while emitting cached signals. */ - private long demandBeforeReadyToWrite; - - /** Current state. */ - private volatile State state = State.NEW; - - /** The actual writeSubscriber from the HTTP server adapter. */ - @Nullable - private Subscriber writeSubscriber; - - - WriteBarrier(CoreSubscriber completionSubscriber) { - this.writeCompletionBarrier = new WriteCompletionBarrier(completionSubscriber, this); - } - - - // Subscriber methods (we're the subscriber to the write source).. - - @Override - public final void onSubscribe(Subscription s) { - if (Operators.validate(this.subscription, s)) { - this.subscription = s; - this.writeCompletionBarrier.connect(); - s.request(1); - } - } - - @Override - public final void onNext(T item) { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onNext(item); - return; - } - //FIXME revisit in case of reentrant sync deadlock - synchronized (this) { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onNext(item); - } - else if (this.state == State.NEW) { - this.item = item; - this.state = State.FIRST_SIGNAL_RECEIVED; - Publisher result; - try { - result = ChannelSendOperator.this.writeFunction.apply(this); - } - catch (Throwable ex) { // NOSONAR - this.writeCompletionBarrier.onError(ex); - return; - } - result.subscribe(this.writeCompletionBarrier); - } - else { - if (this.subscription != null) { - this.subscription.cancel(); - } - this.writeCompletionBarrier.onError(new IllegalStateException("Unexpected item.")); - } - } - } - - private Subscriber requiredWriteSubscriber() { - Subscriber writeSubscriberToReturn = this.writeSubscriber; - Assert.state(writeSubscriberToReturn != null, "No write subscriber"); - return writeSubscriberToReturn; - } - - @Override - public final void onError(Throwable ex) { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onError(ex); - return; - } - synchronized (this) { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onError(ex); - } - else if (this.state == State.NEW) { - this.state = State.FIRST_SIGNAL_RECEIVED; - this.writeCompletionBarrier.onError(ex); - } - else { - this.error = ex; - } - } - } - - @Override - public final void onComplete() { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onComplete(); - return; - } - synchronized (this) { - if (this.state == State.READY_TO_WRITE) { - requiredWriteSubscriber().onComplete(); - } - else if (this.state == State.NEW) { - this.completed = true; - this.state = State.FIRST_SIGNAL_RECEIVED; - Publisher result; - try { - result = ChannelSendOperator.this.writeFunction.apply(this); - } - catch (Throwable ex) { // NOSONAR - this.writeCompletionBarrier.onError(ex); - return; - } - result.subscribe(this.writeCompletionBarrier); - } - else { - this.completed = true; - } - } - } - - @Override - public Context currentContext() { - return this.writeCompletionBarrier.currentContext(); - } - - - // Subscription methods (we're the Subscription to the writeSubscriber).. - - @Override - public void request(long n) { - long requests = n; - Subscription s = this.subscription; - if (s == null) { - return; - } - if (this.state == State.READY_TO_WRITE) { - s.request(requests); - return; - } - synchronized (this) { - if (this.writeSubscriber != null) { - if (this.state == State.EMITTING_CACHED_SIGNALS) { - this.demandBeforeReadyToWrite = requests; - return; - } - try { - this.state = State.EMITTING_CACHED_SIGNALS; - if (emitCachedSignals()) { - return; - } - requests = requests + this.demandBeforeReadyToWrite - 1; - if (requests == 0) { - return; - } - } - finally { - this.state = State.READY_TO_WRITE; - } - } - } - s.request(requests); - } - - private boolean emitCachedSignals() { - if (this.error != null) { - try { - requiredWriteSubscriber().onError(this.error); - } - finally { - releaseCachedItem(); - } - return true; - } - T itemToUse; - synchronized (this) { - itemToUse = this.item; - this.item = null; - } - if (itemToUse != null) { - requiredWriteSubscriber().onNext(itemToUse); - } - if (this.completed) { - requiredWriteSubscriber().onComplete(); - return true; - } - return false; - } - - @Override - public void cancel() { - Subscription s = this.subscription; - if (s != null) { - this.subscription = null; - try { - s.cancel(); - } - finally { - releaseCachedItem(); - } - } - } - - private void releaseCachedItem() { - synchronized (this) { - Object itemToRelease = this.item; - if (itemToRelease instanceof DataBuffer) { - DataBufferUtils.release((DataBuffer) itemToRelease); - } - this.item = null; - } - } - - - // Publisher methods (we're the Publisher to the writeSubscriber).. - - @Override - public void subscribe(Subscriber writeSubscriber) { - synchronized (this) { - Assert.state(this.writeSubscriber == null, "Only one write subscriber supported"); - this.writeSubscriber = writeSubscriber; - if (this.error != null || this.completed) { - this.writeSubscriber.onSubscribe(Operators.emptySubscription()); - emitCachedSignals(); - } - else { - this.writeSubscriber.onSubscribe(this); - } - } - } - - } - - - /** - * We need an extra barrier between the WriteBarrier itself and the actual - * completion subscriber. - * - *

The completionSubscriber is subscribed initially to the WriteBarrier. - * Later after the first signal is received, we need one more subscriber - * instance (per spec can only subscribe once) to subscribe to the write - * function and switch to delegating completion signals from it. - */ - private class WriteCompletionBarrier implements CoreSubscriber, Subscription { - - /* Downstream write completion subscriber */ - private final CoreSubscriber completionSubscriber; - - private final WriteBarrier writeBarrier; - - @Nullable - private Subscription subscription; - - - WriteCompletionBarrier(CoreSubscriber subscriber, WriteBarrier writeBarrier) { - this.completionSubscriber = subscriber; - this.writeBarrier = writeBarrier; - } - - - /** - * Connect the underlying completion subscriber to this barrier in order - * to track cancel signals and pass them on to the write barrier. - */ - void connect() { - this.completionSubscriber.onSubscribe(this); - } - - // Subscriber methods (we're the subscriber to the write function).. - - @Override - public void onSubscribe(Subscription subscription) { - this.subscription = subscription; - subscription.request(Long.MAX_VALUE); - } - - @Override - public void onNext(Void aVoid) { - } - - @Override - public void onError(Throwable ex) { - try { - this.completionSubscriber.onError(ex); - } - finally { - this.writeBarrier.releaseCachedItem(); - } - } - - @Override - public void onComplete() { - this.completionSubscriber.onComplete(); - } - - @Override - public Context currentContext() { - return this.completionSubscriber.currentContext(); - } - - - @Override - public void request(long n) { - // Ignore: we don't produce data - } - - @Override - public void cancel() { - this.writeBarrier.cancel(); - Subscription subscriptionToCancel = this.subscription; - if (subscriptionToCancel != null) { - subscriptionToCancel.cancel(); - } - } - - } - -} diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java index 899c6e927e..cd807c4119 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java @@ -17,6 +17,7 @@ package org.springframework.integration.rsocket.inbound; import java.util.Arrays; +import java.util.concurrent.atomic.AtomicReference; import org.reactivestreams.Publisher; @@ -37,16 +38,13 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler; -import org.springframework.messaging.rsocket.PayloadUtils; import org.springframework.messaging.rsocket.RSocketStrategies; import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler; import org.springframework.util.Assert; import org.springframework.util.MimeType; -import io.rsocket.Payload; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoProcessor; /** * The {@link MessagingGatewaySupport} implementation for the {@link IntegrationRSocketEndpoint}. @@ -203,13 +201,15 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In } Mono> requestMono = decodeRequestMessage(requestMessage); - MonoProcessor> replyMono = getReplyMono(requestMessage); - if (replyMono != null) { + AtomicReference replyTo = getReplyToHeader(requestMessage); + if (replyTo != null) { return requestMono .flatMap(this::sendAndReceiveMessageReactive) - .flatMap((replyMessage) -> - new ChannelSendOperator<>(createReply(replyMessage.getPayload(), requestMessage), - (publisher) -> sendReply(publisher, replyMono))); + .flatMap((replyMessage) -> { + Flux reply = createReply(replyMessage.getPayload(), requestMessage); + replyTo.set(reply); + return Mono.empty(); + }); } else { return requestMono @@ -308,18 +308,12 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In return encoder.encodeValue(element, bufferFactory, elementType, mimeType, null); } - private Mono sendReply(Publisher reply, MonoProcessor> replyMono) { - replyMono.onNext(Flux.from(reply).map(PayloadUtils::createPayload)); - replyMono.onComplete(); - return Mono.empty(); - } - @Nullable @SuppressWarnings("unchecked") - private static MonoProcessor> getReplyMono(Message message) { + private static AtomicReference getReplyToHeader(Message message) { Object headerValue = message.getHeaders().get(RSocketPayloadReturnValueHandler.RESPONSE_HEADER); - Assert.state(headerValue == null || headerValue instanceof MonoProcessor, "Expected MonoProcessor"); - return (MonoProcessor>) headerValue; + Assert.state(headerValue == null || headerValue instanceof AtomicReference, "Expected AtomicReference"); + return (AtomicReference) headerValue; } }