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

Rely on the RSocket FrameType header

Browse Source 
* After introduction an `rsocketFrameType` header in the message,
we don't need a custom `IntegrationRSocket` any more.
Fully rely on the `MessagingRSocket` from the Spring Messaging, only
handle properly a `RSocketFrameTypeMessageCondition.CONNECT_CONDITION`
for connections from clients to emit an `RSocketConnectedEvent`
* Copy a `ChannelSendOperator` from the Spring Messaging for the proper
reply handling, when we are interested in the source `Publisher`
subscription first.
See its JavaDocs for more info
* Reusing already `public PayloadUtils` instead of its method copies
This commit is contained in:
Artem Bilan
2019-07-10 14:47:41 -04:00
parent 53d4faa83c
commit 24348a105c
6 changed files with 527 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java
View File

@@ -53,7 +53,7 @@ public abstract class AbstractRSocketConnector
private MimeType dataMimeType = MimeTypeUtils.TEXT_PLAIN;
private MimeType metadataMimeType = IntegrationRSocket.COMPOSITE_METADATA;
private MimeType metadataMimeType = new MimeType("message", "x.rsocket.composite-metadata.v0");
private RSocketStrategies rsocketStrategies =
RSocketStrategies.builder()

238
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java
View File

@@ -1,238 +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;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import org.reactivestreams.Publisher;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.ReactiveMessageHandler;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.handler.invocation.reactive.HandlerMethodReturnValueHandler;
import org.springframework.messaging.rsocket.PayloadUtils;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.support.MetadataExtractor;
import org.springframework.messaging.rsocket.annotation.support.RSocketPayloadReturnValueHandler;
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.messaging.support.MessageHeaderAccessor;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.RouteMatcher;
import io.rsocket.AbstractRSocket;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.Payload;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
/**
* Implementation of {@link io.rsocket.RSocket} that wraps incoming requests with a
* {@link Message}, delegates to a {@link Function} for handling, and then
* obtains the response from a "reply" header.
* <p>
* Essentially, this is an adapted for Spring Integration copy
* of the {@link org.springframework.messaging.rsocket.annotation.support.MessagingRSocket} because
* that one is not public.
*
* @author Artem Bilan
*
* @since 5.2
*
* @see org.springframework.messaging.rsocket.annotation.support.MessagingRSocket
*/
class IntegrationRSocket extends AbstractRSocket {
static final MimeType COMPOSITE_METADATA = new MimeType("message", "x.rsocket.composite-metadata.v0");
private final ReactiveMessageHandler handler;
private final RouteMatcher routeMatcher;
private final RSocketRequester requester;
private final DataBufferFactory bufferFactory;
private final MimeType dataMimeType;
private final MimeType metadataMimeType;
private final MetadataExtractor metadataExtractor;
IntegrationRSocket(ReactiveMessageHandler handler, RouteMatcher routeMatcher,
RSocketRequester requester, MimeType dataMimeType, MimeType metadataMimeType,
MetadataExtractor metadataExtractor, DataBufferFactory bufferFactory) {
Assert.notNull(handler, "'handler' is required");
Assert.notNull(routeMatcher, "'routeMatcher' is required");
Assert.notNull(requester, "'requester' is required");
Assert.notNull(dataMimeType, "'dataMimeType' is required");
Assert.notNull(metadataMimeType, "'metadataMimeType' is required");
this.handler = handler;
this.routeMatcher = routeMatcher;
this.requester = requester;
this.dataMimeType = dataMimeType;
this.metadataMimeType = metadataMimeType;
this.metadataExtractor = metadataExtractor;
this.bufferFactory = bufferFactory;
}
RSocketRequester getRequester() {
return this.requester;
}
/**
* Wrap the {@link ConnectionSetupPayload} with a {@link Message} and
* delegate to {@link #handle(Payload)} for handling.
* @param payload the connection payload
* @return completion handle for success or error
*/
Mono<DataBuffer> handleConnectionSetupPayload(ConnectionSetupPayload payload) {
DataBuffer dataBuffer = retainDataAndReleasePayload(payload);
int refCount = refCount(dataBuffer);
return Mono.just(dataBuffer)
.doFinally(s -> {
if (refCount(dataBuffer) == refCount) {
DataBufferUtils.release(dataBuffer);
}
});
}
@Override
public Mono<Void> fireAndForget(Payload payload) {
return handle(payload);
}
@Override
public Mono<Payload> requestResponse(Payload payload) {
return handleAndReply(payload, Flux.just(payload)).next();
}
@Override
public Flux<Payload> requestStream(Payload payload) {
return handleAndReply(payload, Flux.just(payload));
}
@Override
public Flux<Payload> requestChannel(Publisher<Payload> payloads) {
return Flux.from(payloads)
.switchOnFirst((signal, innerFlux) -> {
Payload firstPayload = signal.get();
return firstPayload == null ? innerFlux : handleAndReply(firstPayload, innerFlux);
});
}
@Override
public Mono<Void> metadataPush(Payload payload) {
// Not very useful until createHeaders does more with metadata
return handle(payload);
}
private Mono<Void> handle(Payload payload) {
MessageHeaders headers = createHeaders(payload, null);
DataBuffer dataBuffer = retainDataAndReleasePayload(payload);
int refCount = refCount(dataBuffer);
Message<?> message = MessageBuilder.createMessage(dataBuffer, headers);
return Mono.defer(() -> this.handler.handleMessage(message))
.doFinally((signal) -> {
if (refCount(dataBuffer) == refCount) {
DataBufferUtils.release(dataBuffer);
}
});
}
private Flux<Payload> handleAndReply(Payload firstPayload, Flux<Payload> payloads) {
MonoProcessor<Flux<Payload>> replyMono = MonoProcessor.create();
MessageHeaders headers = createHeaders(firstPayload, replyMono);
AtomicBoolean read = new AtomicBoolean();
Flux<DataBuffer> buffers =
payloads.map(this::retainDataAndReleasePayload)
.doOnSubscribe((subscription) -> read.set(true));
Message<Flux<DataBuffer>> message = MessageBuilder.createMessage(buffers, headers);
return Mono.defer(() -> this.handler.handleMessage(message))
.doFinally((signal) -> {
// Subscription should have happened by now due to ChannelSendOperator
if (!read.get()) {
buffers.subscribe(DataBufferUtils::release);
}
})
.thenMany(Flux.defer(() ->
replyMono.isTerminated()
? replyMono.flatMapMany(Function.identity())
: Mono.error(new IllegalStateException("Something went wrong: reply Mono not set"))));
}
String getDestination(Payload payload) {
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
Object routingKey = metadataValues.get(MetadataExtractor.ROUTE_KEY);
if (routingKey != null) {
RouteMatcher.Route route = this.routeMatcher.parseRoute(routingKey.toString());
return route.value();
}
else {
return "";
}
}
private DataBuffer retainDataAndReleasePayload(Payload payload) {
payload.retain();
return PayloadUtils.retainDataAndReleasePayload(payload, this.bufferFactory);
}
private MessageHeaders createHeaders(Payload payload, @Nullable MonoProcessor<?> replyMono) {
MessageHeaderAccessor headers = new MessageHeaderAccessor();
headers.setLeaveMutable(true);
Map<String, Object> metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType);
metadataValues.putIfAbsent(MetadataExtractor.ROUTE_KEY, "");
for (Map.Entry<String, Object> entry : metadataValues.entrySet()) {
if (entry.getKey().equals(MetadataExtractor.ROUTE_KEY)) {
RouteMatcher.Route route = this.routeMatcher.parseRoute((String) entry.getValue());
headers.setHeader(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER, route);
}
else {
headers.setHeader(entry.getKey(), entry.getValue());
}
}
headers.setContentType(this.dataMimeType);
headers.setHeader(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER, this.requester);
headers.setHeader(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER, this.bufferFactory);
headers.setHeader(RSocketPayloadReturnValueHandler.RESPONSE_HEADER, replyMono);
return headers.getMessageHeaders();
}
private static int refCount(DataBuffer dataBuffer) {
return dataBuffer instanceof NettyDataBuffer ? ((NettyDataBuffer) dataBuffer).getNativeBuffer().refCnt() : 1;
}
}

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

@@ -19,30 +19,18 @@ package org.springframework.integration.rsocket;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
import java.util.function.BiFunction;
import org.springframework.context.ApplicationContext;
import org.springframework.core.MethodParameter;
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.SyncHandlerMethodArgumentResolver;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.DefaultMetadataExtractor;
import org.springframework.messaging.rsocket.annotation.support.MetadataExtractor;
import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import io.rsocket.ConnectionSetupPayload;
import io.rsocket.RSocket;
/**
* The {@link RSocketMessageHandler} extension for Spring Integration needs.
@@ -60,63 +48,10 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
private static final Method HANDLE_MESSAGE_METHOD =
ReflectionUtils.findMethod(ReactiveMessageHandler.class, "handleMessage", Message.class);
@Nullable
private MimeType defaultDataMimeType;
private MimeType defaultMetadataMimeType = IntegrationRSocket.COMPOSITE_METADATA;
private MetadataExtractor metadataExtractor;
IntegrationRSocketMessageHandler() {
setHandlerPredicate((clazz) -> false);
}
/**
* Configure the default content type to use for data payloads.
* <p>By default this is not set. However a server acceptor will use the
* content type from the {@link io.rsocket.ConnectionSetupPayload}, so this is typically
* required for clients but can also be used on servers as a fallback.
* @param defaultDataMimeType the MimeType to use
*/
@Override
public void setDefaultDataMimeType(@Nullable MimeType defaultDataMimeType) {
super.setDefaultDataMimeType(defaultDataMimeType);
this.defaultDataMimeType = defaultDataMimeType;
}
/**
* Configure the default {@code MimeType} for payload data if the
* {@code SETUP} frame did not specify one.
* <p>By default this is set to {@code "message/x.rsocket.composite-metadata.v0"}
* @param mimeType the MimeType to use
*/
@Override
public void setDefaultMetadataMimeType(MimeType mimeType) {
super.setDefaultMetadataMimeType(mimeType);
this.defaultMetadataMimeType = mimeType;
}
/**
* Configure a {@link MetadataExtractor} to extract the route and possibly
* other metadata from the first payload of incoming requests.
* <p>By default this is a {@link DefaultMetadataExtractor} with the
* configured {@link RSocketStrategies} (and decoders), extracting a route
* from {@code "message/x.rsocket.routing.v0"} or {@code "text/plain"}
* metadata entries.
* @param extractor the extractor to use
*/
@Override
public void setMetadataExtractor(MetadataExtractor extractor) {
super.setMetadataExtractor(extractor);
this.metadataExtractor = extractor;
}
@Override
public BiFunction<ConnectionSetupPayload, RSocket, RSocket> clientAcceptor() {
return this::createRSocket;
}
public boolean detectEndpoints() {
ApplicationContext applicationContext = getApplicationContext();
if (applicationContext != null && getHandlerMethods().isEmpty()) {
@@ -135,6 +70,7 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
public void addEndpoint(IntegrationRSocketEndpoint endpoint) {
registerHandlerMethod(endpoint, HANDLE_MESSAGE_METHOD,
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.REQUEST_CONDITION,
new DestinationPatternsMessageCondition(endpoint.getPath(), getRouteMatcher())));
}
@@ -143,36 +79,6 @@ class IntegrationRSocketMessageHandler extends RSocketMessageHandler {
return Collections.singletonList(new MessageHandlerMethodArgumentResolver());
}
@Override
public void afterPropertiesSet() {
super.afterPropertiesSet();
if (this.metadataExtractor == null) {
DefaultMetadataExtractor extractor = new DefaultMetadataExtractor(getRSocketStrategies()); // NOSONAR
extractor.metadataToExtract(MimeTypeUtils.TEXT_PLAIN, String.class, MetadataExtractor.ROUTE_KEY);
this.metadataExtractor = extractor;
}
}
protected IntegrationRSocket createRSocket(ConnectionSetupPayload setupPayload, RSocket rsocket) {
String mimeType = setupPayload.dataMimeType();
MimeType dataMimeType =
StringUtils.hasText(mimeType)
? MimeTypeUtils.parseMimeType(mimeType)
: this.defaultDataMimeType;
Assert.notNull(dataMimeType, "No `dataMimeType` in ConnectionSetupPayload and no default value");
mimeType = setupPayload.metadataMimeType();
MimeType metaMimeType =
StringUtils.hasText(mimeType)
? MimeTypeUtils.parseMimeType(mimeType)
: this.defaultMetadataMimeType;
Assert.notNull(dataMimeType, "No `metadataMimeType` in ConnectionSetupPayload and no default value");
RSocketStrategies rSocketStrategies = getRSocketStrategies();
Assert.notNull(rSocketStrategies, "No `rSocketStrategies` provided");
RSocketRequester requester = RSocketRequester.wrap(rsocket, dataMimeType, metaMimeType, rSocketStrategies);
return new IntegrationRSocket(this, getRouteMatcher(), requester, dataMimeType, metaMimeType,
this.metadataExtractor, rSocketStrategies.dataBufferFactory());
}
private static final class MessageHandlerMethodArgumentResolver implements SyncHandlerMethodArgumentResolver {
@Override

77
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java
View File

@@ -22,15 +22,20 @@ import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.CompositeMessageCondition;
import org.springframework.messaging.handler.DestinationPatternsMessageCondition;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.support.RSocketFrameTypeMessageCondition;
import org.springframework.messaging.rsocket.annotation.support.RSocketRequesterMethodArgumentResolver;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.RouteMatcher;
import io.rsocket.RSocketFactory;
import io.rsocket.SocketAcceptor;
@@ -162,9 +167,13 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
.subscribe();
}
private static class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler {
@Override
public void afterSingletonsInstantiated() {
super.afterSingletonsInstantiated();
serverRSocketMessageHandler().registerHandleConnectionSetupMethod();
}
private static final Log LOGGER = LogFactory.getLog(ServerRSocketMessageHandler.class);
private static class ServerRSocketMessageHandler extends IntegrationRSocketMessageHandler {
private final Map<Object, RSocketRequester> clientRSocketRequesters = new HashMap<>();
@@ -172,29 +181,41 @@ public class ServerRSocketConnector extends AbstractRSocketConnector
private ApplicationEventPublisher applicationEventPublisher;
@Override
public SocketAcceptor serverAcceptor() {
return (setupPayload, sendingRSocket) -> {
IntegrationRSocket rsocket = createRSocket(setupPayload, sendingRSocket);
return rsocket.handleConnectionSetupPayload(setupPayload)
.doOnNext((dataBuffer) -> {
String destination = rsocket.getDestination(setupPayload);
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer);
RSocketRequester rsocketRequester = rsocket.getRequester();
this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
RSocketConnectedEvent rSocketConnectedEvent =
new RSocketConnectedEvent(rsocket, destination, dataBuffer, rsocketRequester);
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
}
else {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("The RSocket has been connected: " + rSocketConnectedEvent);
}
}
})
.thenReturn(rsocket);
};
private void registerHandleConnectionSetupMethod() {
registerHandlerMethod(this,
ReflectionUtils.findMethod(ServerRSocketMessageHandler.class, "handleConnectionSetup", // NOSONAR
Message.class),
new CompositeMessageCondition(
RSocketFrameTypeMessageCondition.CONNECT_CONDITION,
new DestinationPatternsMessageCondition(new String[] { "*" }, getRouteMatcher())));
}
private void handleConnectionSetup(Message<DataBuffer> connectMessage) {
DataBuffer dataBuffer = connectMessage.getPayload();
MessageHeaders messageHeaders = connectMessage.getHeaders();
String destination = "";
RouteMatcher.Route route =
messageHeaders.get(DestinationPatternsMessageCondition.LOOKUP_DESTINATION_HEADER,
RouteMatcher.Route.class);
if (route != null) {
destination = route.value();
}
Object rsocketRequesterKey = this.clientRSocketKeyStrategy.apply(destination, dataBuffer);
RSocketRequester rsocketRequester =
messageHeaders.get(RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER,
RSocketRequester.class);
this.clientRSocketRequesters.put(rsocketRequesterKey, rsocketRequester);
RSocketConnectedEvent rSocketConnectedEvent =
new RSocketConnectedEvent(this, destination, dataBuffer, rsocketRequester); // NOSONAR
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(rSocketConnectedEvent);
}
else {
if (logger.isInfoEnabled()) {
logger.info("The RSocket has been connected: " + rSocketConnectedEvent);
}
}
}
}

462
spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java Normal file
View File

@@ -0,0 +1,462 @@
/*
* 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 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) {
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() {
Assert.state(this.writeSubscriber != null, "No write subscriber");
return this.writeSubscriber;
}
@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) {
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) {
Subscription s = this.subscription;
if (s == null) {
return;
}
if (this.state == State.READY_TO_WRITE) {
s.request(n);
return;
}
synchronized (this) {
if (this.writeSubscriber != null) {
if (this.state == State.EMITTING_CACHED_SIGNALS) {
this.demandBeforeReadyToWrite = n;
return;
}
try {
this.state = State.EMITTING_CACHED_SIGNALS;
if (emitCachedSignals()) {
return;
}
n = n + this.demandBeforeReadyToWrite - 1;
if (n == 0) {
return;
}
}
finally {
this.state = State.READY_TO_WRITE;
}
}
}
s.request(n);
}
private boolean emitCachedSignals() {
if (this.error != null) {
try {
requiredWriteSubscriber().onError(this.error);
}
finally {
releaseCachedItem();
}
return true;
}
T item = this.item;
this.item = null;
if (item != null) {
requiredWriteSubscriber().onNext(item);
}
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 item = this.item;
if (item instanceof DataBuffer) {
DataBufferUtils.release((DataBuffer) item);
}
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 subscription = this.subscription;
if (subscription != null) {
subscription.cancel();
}
}
}
}

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

@@ -28,9 +28,7 @@ import org.springframework.core.codec.Encoder;
import org.springframework.core.codec.StringDecoder;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DefaultDataBuffer;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.rsocket.AbstractRSocketConnector;
import org.springframework.integration.rsocket.ClientRSocketConnector;
@@ -41,14 +39,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 io.rsocket.util.ByteBufPayload;
import io.rsocket.util.DefaultPayload;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoProcessor;
@@ -184,11 +181,9 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
if (replyMono != null) {
return requestMono
.flatMap(this::sendAndReceiveMessageReactive)
.doOnNext(replyMessage -> {
replyMono.onNext(createReply(replyMessage.getPayload(), requestMessage));
replyMono.onComplete();
})
.then();
.flatMap((replyMessage) ->
new ChannelSendOperator<>(createReply(replyMessage.getPayload(), requestMessage),
(publisher) -> sendReply(publisher, replyMono)));
}
else {
return requestMono
@@ -218,7 +213,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
Object payload = requestMessage.getPayload();
// The IntegrationRSocket logic ensures that we can have only a single DataBuffer payload or Flux<DataBuffer>.
// The MessagingRSocket logic ensures that we can have only a single DataBuffer payload or Flux<DataBuffer>.
Decoder<Object> decoder = this.rsocketStrategies.decoder(elementType, mimeType);
if (payload instanceof DataBuffer) {
return decoder.decode((DataBuffer) payload, elementType, mimeType, null);
@@ -228,7 +223,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
}
}
private Flux<Payload> createReply(Object reply, Message<?> requestMessage) {
private Flux<DataBuffer> createReply(Object reply, Message<?> requestMessage) {
MessageHeaders requestMessageHeaders = requestMessage.getHeaders();
DataBufferFactory bufferFactory =
requestMessageHeaders.get(HandlerMethodReturnValueHandler.DATA_BUFFER_FACTORY_HEADER,
@@ -240,8 +235,7 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
MimeType mimeType = requestMessageHeaders.get(MessageHeaders.CONTENT_TYPE, MimeType.class);
return encodeContent(reply, ResolvableType.forInstance(reply), bufferFactory, mimeType)
.map(RSocketInboundGateway::createPayload);
return encodeContent(reply, ResolvableType.forInstance(reply), bufferFactory, mimeType);
}
private Flux<DataBuffer> encodeContent(Object content, ResolvableType returnValueType,
@@ -269,6 +263,12 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
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
@SuppressWarnings("unchecked")
private static MonoProcessor<Flux<Payload>> getReplyMono(Message<?> message) {
@@ -277,16 +277,4 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In
return (MonoProcessor<Flux<Payload>>) headerValue;
}
private static Payload createPayload(DataBuffer data) {
if (data instanceof NettyDataBuffer) {
return ByteBufPayload.create(((NettyDataBuffer) data).getNativeBuffer());
}
else if (data instanceof DefaultDataBuffer) {
return DefaultPayload.create(((DefaultDataBuffer) data).getNativeBuffer());
}
else {
return DefaultPayload.create(data.asByteBuffer());
}
}
}