diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java index 7f4afa1388..35aa5dd705 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/AbstractRSocketConnector.java @@ -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() diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java deleted file mode 100644 index 39d773099b..0000000000 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/IntegrationRSocket.java +++ /dev/null @@ -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. - *

- * 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 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 fireAndForget(Payload payload) { - return handle(payload); - } - - @Override - public Mono requestResponse(Payload payload) { - return handleAndReply(payload, Flux.just(payload)).next(); - } - - @Override - public Flux requestStream(Payload payload) { - return handleAndReply(payload, Flux.just(payload)); - } - - @Override - public Flux requestChannel(Publisher payloads) { - return Flux.from(payloads) - .switchOnFirst((signal, innerFlux) -> { - Payload firstPayload = signal.get(); - return firstPayload == null ? innerFlux : handleAndReply(firstPayload, innerFlux); - }); - } - - @Override - public Mono metadataPush(Payload payload) { - // Not very useful until createHeaders does more with metadata - return handle(payload); - } - - - private Mono 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 handleAndReply(Payload firstPayload, Flux payloads) { - MonoProcessor> replyMono = MonoProcessor.create(); - MessageHeaders headers = createHeaders(firstPayload, replyMono); - - AtomicBoolean read = new AtomicBoolean(); - Flux buffers = - payloads.map(this::retainDataAndReleasePayload) - .doOnSubscribe((subscription) -> read.set(true)); - Message> 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 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 metadataValues = this.metadataExtractor.extract(payload, this.metadataMimeType); - metadataValues.putIfAbsent(MetadataExtractor.ROUTE_KEY, ""); - for (Map.Entry 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; - } - -} 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 c57e09ae8c..797f1466c3 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 @@ -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. - *

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. - *

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. - *

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 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 diff --git a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java index 31e6963a48..e89685f4ac 100644 --- a/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/ServerRSocketConnector.java @@ -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 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 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); + } + } } } 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 new file mode 100644 index 0000000000..fd97256c4c --- /dev/null +++ b/spring-integration-rsocket/src/main/java/org/springframework/integration/rsocket/inbound/ChannelSendOperator.java @@ -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; + +/** + * ---------------------- + *

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: + *

+ * + *

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 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) { + 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() { + 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 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 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 subscription = this.subscription; + if (subscription != null) { + subscription.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 0801b00621..7d169c89df 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 @@ -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. + // The MessagingRSocket logic ensures that we can have only a single DataBuffer payload or Flux. Decoder 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 createReply(Object reply, Message requestMessage) { + private Flux 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 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 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) { @@ -277,16 +277,4 @@ public class RSocketInboundGateway extends MessagingGatewaySupport implements In return (MonoProcessor>) 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()); - } - } - }