Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Upgrade to the latest Reactor & SF

Browse Source 
Related to https://github.com/spring-projects/spring-framework/issues/25884

* Don't use `MonoProcessor` & `FluxProcessor` since they are deprecated now
* Fix RSocket components to use an `AtomicReference` header instead of deprecated
`MonoProcessor`
* Introduce an internal `IntegrationRSocketPayloadReturnValueHandler` to handle
properly a Spring Integration case for inbound requests with its
`IntegrationRSocketEndpoint` implementation
* Don't do response handling in the `RSocketInboundGateway` any more
since it is deferred now to the `IntegrationRSocketPayloadReturnValueHandler`
* Remove internal `ChannelSendOperator` since it is out of use from now on
This commit is contained in:
Artem Bilan
2020-10-12 11:24:39 -04:00
parent 20447df7a2
commit cadd4e6b7f
6 changed files with 83 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/channel/FluxMessageChannel.java
View File

@@ -29,7 +29,6 @@ import org.springframework.util.Assert;
import reactor.core.Disposable; import reactor.core.Disposable;
import reactor.core.Disposables; import reactor.core.Disposables;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers; import reactor.core.scheduler.Schedulers;
@@ -46,9 +45,7 @@ import reactor.core.scheduler.Schedulers;
public class FluxMessageChannel extends AbstractMessageChannel public class FluxMessageChannel extends AbstractMessageChannel
implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel { implements Publisher<Message<?>>, ReactiveStreamsSubscribableChannel {
private final Sinks.Many<Message<?>> sink; private final Sinks.Many<Message<?>> sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
private final FluxProcessor<Message<?>, Message<?>> processor;
private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1); private final Sinks.Many<Boolean> subscribedSignal = Sinks.many().replay().limit(1);
@@ -56,14 +53,9 @@ public class FluxMessageChannel extends AbstractMessageChannel
private volatile boolean active = true; private volatile boolean active = true;
public FluxMessageChannel() {
this.sink = Sinks.many().multicast().onBackpressureBuffer(1, false);
this.processor = FluxProcessor.fromSink(this.sink);
}
@Override @Override
protected boolean doSend(Message<?> message, long timeout) { 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"); () -> "The [" + this + "] doesn't have subscribers to accept messages");
long remainingTime = 0; long remainingTime = 0;
if (timeout > 0) { if (timeout > 0) {
@@ -101,11 +93,12 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override @Override
public void subscribe(Subscriber<? super Message<?>> subscriber) { public void subscribe(Subscriber<? super Message<?>> subscriber) {
this.processor this.sink.asFlux()
.doFinally((s) -> this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams())) .doFinally((s) -> this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0))
.share() .share()
.subscribe(subscriber); .subscribe(subscriber);
this.subscribedSignal.tryEmitNext(this.processor.hasDownstreams());
this.subscribedSignal.tryEmitNext(this.sink.currentSubscriberCount() > 0);
} }
@Override @Override
@@ -131,9 +124,9 @@ public class FluxMessageChannel extends AbstractMessageChannel
@Override @Override
public void destroy() { public void destroy() {
this.active = false; this.active = false;
this.subscribedSignal.tryEmitNext(false);
this.upstreamSubscriptions.dispose(); this.upstreamSubscriptions.dispose();
this.processor.onComplete(); this.subscribedSignal.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
this.sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
super.destroy(); super.destroy();
} }

6
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -67,7 +67,6 @@ import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
import reactor.core.publisher.Sinks; import reactor.core.publisher.Sinks;
/** /**
@@ -899,7 +898,10 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint
@Override @Override
public void subscribeTo(Publisher<? extends Message<?>> publisher) { public void subscribeTo(Publisher<? extends Message<?>> 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);
} }
} }

3
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.java
View File

@@ -51,7 +51,6 @@ import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import reactor.core.Disposable; import reactor.core.Disposable;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;
/** /**
* @author Artem Bilan * @author Artem Bilan
@@ -141,7 +140,7 @@ public class FluxMessageChannelTests {
flowRegistration.destroy(); flowRegistration.destroy();
assertThat(TestUtils.getPropertyValue(flux, "processor", FluxProcessor.class).isTerminated()).isTrue(); assertThat(TestUtils.getPropertyValue(flux, "sink.sink.done", Boolean.class)).isTrue();
} }
@Configuration @Configuration

59
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocketMessageHandler.java
View File

@@ -17,24 +17,35 @@
package org.springframework.integration.rsocket; package org.springframework.integration.rsocket;
import java.lang.reflect.Method; import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collection; import java.util.Collection;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContext;
import org.springframework.core.MethodParameter; 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.Message;
import org.springframework.messaging.ReactiveMessageHandler; import org.springframework.messaging.ReactiveMessageHandler;
import org.springframework.messaging.handler.CompositeMessageCondition; import org.springframework.messaging.handler.CompositeMessageCondition;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition; import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.handler.invocation.reactive.HandlerMethodArgumentResolver; 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.handler.invocation.reactive.SyncHandlerMethodArgumentResolver;
import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition; import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler; 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 org.springframework.util.ReflectionUtils;
import io.rsocket.Payload;
import io.rsocket.frame.FrameType; import io.rsocket.frame.FrameType;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
/** /**
* The {@link RSocketMessageHandler} extension for Spring Integration needs. * The {@link RSocketMessageHandler} extension for Spring Integration needs.
@@ -109,6 +120,23 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
} }
} }
@Override
@SuppressWarnings("unchecked")
protected List<? extends HandlerMethodReturnValueHandler> initReturnValueHandlers() {
HandlerMethodReturnValueHandler integrationRSocketPayloadReturnValueHandler =
new IntegrationRSocketPayloadReturnValueHandler((List<Encoder<?>>) getEncoders(),
getReactiveAdapterRegistry());
if (this.messageMappingCompatible) {
List<HandlerMethodReturnValueHandler> handlers = new ArrayList<>();
handlers.add(integrationRSocketPayloadReturnValueHandler);
handlers.addAll(getReturnValueHandlerConfigurer().getCustomHandlers());
return handlers;
}
else {
return Collections.singletonList(integrationRSocketPayloadReturnValueHandler);
}
}
protected static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver { protected static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
@Override @Override
@@ -123,4 +151,35 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
} }
protected static final class IntegrationRSocketPayloadReturnValueHandler extends RSocketPayloadReturnValueHandler {
protected IntegrationRSocketPayloadReturnValueHandler(List<Encoder<?>> encoders,
ReactiveAdapterRegistry registry) {
super(encoders, registry);
}
@Override public Mono<Void> handleReturnValue(@Nullable Object returnValue, MethodParameter returnType,
Message<?> message) {
AtomicReference<Flux<Payload>> 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<Flux<Payload>> getResponseReference(Message<?> message) {
Object headerValue = message.getHeaders().get(RESPONSE_HEADER);
Assert.state(headerValue == null || headerValue instanceof AtomicReference, "Expected AtomicReference");
return (AtomicReference<Flux<Payload>>) headerValue;
}
}
} }

467
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java
View File

@@ -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;
/**
* ----------------------
* <p><strong>NOTE:</strong> 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.
* <p>----------------------<br>
*
* <p>Given a write function that accepts a source {@code Publisher<T>} to write
* with and returns {@code Publisher<Void>} 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 <T> the type of element signaled
*/
class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
private final Function<Publisher<T>, Publisher<Void>> writeFunction;
private final Flux<T> source;
ChannelSendOperator(Publisher<? extends T> source, Function<Publisher<T>, Publisher<Void>> 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<? super Void> 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.
*
* <p>Acts as:
* <ul>
* <li>Subscriber to the write source.
* <li>Subscription to the write subscriber.
* <li>Publisher to the write subscriber.
* </ul>
*
* <p>Also uses {@link WriteCompletionBarrier} to communicate completion
* and detect cancel signals from the completion subscriber.
*/
private class WriteBarrier implements CoreSubscriber<T>, Subscription, Publisher<T> {
/* 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<? super T> writeSubscriber;
WriteBarrier(CoreSubscriber<? super Void> completionSubscriber) {
this.writeCompletionBarrier = new WriteCompletionBarrier(completionSubscriber, this);
}
// Subscriber<T> 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<Void> 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<? super T> requiredWriteSubscriber() {
Subscriber<? super T> 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<Void> 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<T> methods (we're the Publisher to the writeSubscriber)..
@Override
public void subscribe(Subscriber<? super T> 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.
*
* <p>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<Void>, Subscription {
/* Downstream write completion subscriber */
private final CoreSubscriber<? super Void> completionSubscriber;
private final WriteBarrier writeBarrier;
@Nullable
private Subscription subscription;
WriteCompletionBarrier(CoreSubscriber<? super Void> 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();
}
}
}
}

28
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/RSocketInboundGateway.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.rsocket.inbound; package org.springframework.integration.rsocket.inbound;
import java.util.Arrays; import java.util.Arrays;
import java.util.concurrent.atomic.AtomicReference;
import org.reactivestreams.Publisher; import org.reactivestreams.Publisher;
@@ -37,16 +38,13 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler; 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.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler; import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.util.MimeType; import org.springframework.util.MimeType;
import io.rsocket.Payload;
import reactor.core.publisher.Flux; import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono; import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
/** /**
* The {@link MessagingGatewaySupport} implementation for the {@link IntegrationRSocketEndpoint}. * The {@link MessagingGatewaySupport} implementation for the {@link IntegrationRSocketEndpoint}.
@@ -203,13 +201,15 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
} }
Mono<Message<?>> requestMono = decodeRequestMessage(requestMessage); Mono<Message<?>> requestMono = decodeRequestMessage(requestMessage);
MonoProcessor<Flux<Payload>> replyMono = getReplyMono(requestMessage); AtomicReference<Object> replyTo = getReplyToHeader(requestMessage);
if (replyMono != null) { if (replyTo != null) {
return requestMono return requestMono
.flatMap(this::sendAndReceiveMessageReactive) .flatMap(this::sendAndReceiveMessageReactive)
.flatMap((replyMessage) -> .flatMap((replyMessage) -> {
new ChannelSendOperator<>(createReply(replyMessage.getPayload(), requestMessage), Flux<DataBuffer> reply = createReply(replyMessage.getPayload(), requestMessage);
(publisher) -> sendReply(publisher, replyMono))); replyTo.set(reply);
return Mono.empty();
});
} }
else { else {
return requestMono return requestMono
@@ -308,18 +308,12 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
return encoder.encodeValue(element, bufferFactory, elementType, mimeType, null); return encoder.encodeValue(element, bufferFactory, elementType, mimeType, null);
} }
private Mono<Void> sendReply(Publisher<DataBuffer> reply, MonoProcessor<Flux<Payload>> replyMono) {
replyMono.onNext(Flux.from(reply).map(PayloadUtils::createPayload));
replyMono.onComplete();
return Mono.empty();
}
@Nullable @Nullable
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private static MonoProcessor<Flux<Payload>> getReplyMono(Message<?> message) { private static AtomicReference<Object> getReplyToHeader(Message<?> message) {
Object headerValue = message.getHeaders().get(RSocketPayloadReturnValueHandler.RESPONSE_HEADER); Object headerValue = message.getHeaders().get(RSocketPayloadReturnValueHandler.RESPONSE_HEADER);
Assert.state(headerValue == null || headerValue instanceof MonoProcessor, "Expected MonoProcessor"); Assert.state(headerValue == null || headerValue instanceof AtomicReference, "Expected AtomicReference");
return (MonoProcessor<Flux<Payload>>) headerValue; return (AtomicReference<Object>) headerValue;
} }
} }