org.springframework.integration.rsocket.outbound

Class RSocketOutboundGateway

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.integration.handler.AbstractReplyProducingMessageHandler

org.springframework.integration.rsocket.outbound.RSocketOutboundGateway

All Implemented Interfaces:

org.reactivestreams.Subscriber<Message<?>>, Aware, BeanClassLoaderAware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, Ordered, ExpressionCapable, Orderable, MessageProducer, HeaderPropagationAware, NamedComponent, BaseHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, IntegrationManagement, IntegrationStatsManagement, MessageHandlerMetrics, TrackableComponent, MessageHandler, reactor.core.CoreSubscriber<Message<?>>

public class RSocketOutboundGateway
extends AbstractReplyProducingMessageHandler

An Outbound Messaging Gateway for RSocket requests. The request logic is fully based on the RSocketRequester, which can be obtained from the ClientRSocketConnector on the client side or from the RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER request message header on the server side.

An RSocket operation is determined by the configured RSocketOutboundGateway.Command or respective SpEL expression to be evaluated at runtime against the request message. By default the RSocketOutboundGateway.Command.requestResponse operation is used.

For a Publisher-based requests, it must be present in the request message payload. The flattening via upstream FluxMessageChannel will work, too, but this way we will lose a scope of particular request and every Publisher event will be send in its own plain request.

If reply is a Flux, it is wrapped to the Mono to retain a request scope. The downstream flow is responsible to obtain this Flux from a message payload and subscribe to it by itself. The Mono reply from this component is subscribed from the downstream FluxMessageChannel or it is adapted to the ListenableFuture otherwise.

Since:

5.2

Author:

Artem Bilan

See Also:

RSocketOutboundGateway.Command, RSocketRequester

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	RSocketOutboundGateway.Command Enumeration of commands supported by the gateways.

Nested classes/interfaces inherited from class org.springframework.integration.handler.AbstractReplyProducingMessageHandler

AbstractReplyProducingMessageHandler.RequestHandler

Nested classes/interfaces inherited from interface org.springframework.integration.support.management.IntegrationManagement

IntegrationManagement.ManagementOverrides

Field Summary

Fields inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

messagingTemplate

Fields inherited from class org.springframework.integration.context.IntegrationObjectSupport

EXPRESSION_PARSER, logger

Fields inherited from interface org.springframework.integration.support.management.IntegrationManagement

METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

Fields inherited from interface org.springframework.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

Constructor Summary

Constructors

Constructor and Description
RSocketOutboundGateway(Expression routeExpression) Instantiate based on the provided SpEL expression to evaluate an RSocket endpoint route at runtime against a request message.
RSocketOutboundGateway(String route) Instantiate based on the provided RSocket endpoint route.

Method Summary

Modifier and Type	Method and Description
protected void	doInit()
protected Object	handleRequestMessage(Message<?> requestMessage) Subclasses must implement this method to handle the request Message.
void	setClientRSocketConnector(ClientRSocketConnector clientRSocketConnector) Configure a ClientRSocketConnector for client side requests based on the connection provided by the ClientRSocketConnector.getRSocketRequester().
void	setCommand(RSocketOutboundGateway.Command command) Configure a RSocketOutboundGateway.Command for RSocket request type.
void	setCommandExpression(Expression commandExpression) Configure a SpEL expression to evaluate a RSocketOutboundGateway.Command for RSocket request type at runtime against a request message.
void	setExpectedResponseType(Class<?> expectedResponseType) Specify the expected response type for the RSocket response.
void	setExpectedResponseTypeExpression(Expression expectedResponseTypeExpression) Specify the Expression to determine the type for the RSocket response.
void	setPublisherElementType(Class<?> publisherElementType) Configure a type for a request Publisher elements.
void	setPublisherElementTypeExpression(Expression publisherElementTypeExpression) Configure a SpEL expression to evaluate a request Publisher elements type at runtime against a request message.

Methods inherited from class org.springframework.integration.handler.AbstractReplyProducingMessageHandler

doInvokeAdvisedRequestHandler, getBeanClassLoader, getRequiresReply, handleMessageInternal, hasAdviceChain, onInit, setAdviceChain, setBeanClassLoader, setRequiresReply

Methods inherited from class org.springframework.integration.handler.AbstractMessageProducingHandler

addNotPropagatedHeaders, createOutputMessage, getNotPropagatedHeaders, getOutputChannel, isAsync, produceOutput, resolveErrorChannel, sendErrorMessage, sendOutput, sendOutputs, setAsync, setNotPropagatedHeaders, setOutputChannel, setOutputChannelName, setSendTimeout, shouldCopyRequestHeaders, shouldSplitOutput, updateNotPropagatedHeaders

Methods inherited from class org.springframework.integration.handler.AbstractMessageHandler

configureMetrics, destroy, getActiveCount, getActiveCountLong, getComponentType, getDuration, getErrorCount, getErrorCountLong, getHandleCount, getHandleCountLong, getManagedName, getManagedType, getMaxDuration, getMeanDuration, getMinDuration, getOrder, getOverrides, getStandardDeviationDuration, handleMessage, isCountsEnabled, isLoggingEnabled, isStatsEnabled, onComplete, onError, onNext, onSubscribe, registerMetricsCaptor, reset, setCountsEnabled, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, setStatsEnabled

Methods inherited from class org.springframework.integration.context.IntegrationObjectSupport

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanFactory, getBeanName, getChannelResolver, getComponentName, getConversionService, getExpression, getIntegrationProperties, getIntegrationProperty, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentName, setConversionService, setMessageBuilderFactory, setPrimaryExpression, setTaskScheduler, toString

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Methods inherited from interface org.springframework.integration.support.management.MessageHandlerMetrics

errorCount, handleCount

Methods inherited from interface org.springframework.integration.support.context.NamedComponent

getBeanName, getComponentName

Methods inherited from interface reactor.core.CoreSubscriber

currentContext

Constructor Detail

RSocketOutboundGateway

public RSocketOutboundGateway(String route)

Instantiate based on the provided RSocket endpoint route.

Parameters:

route - the RSocket endpoint route to use.

RSocketOutboundGateway

public RSocketOutboundGateway(Expression routeExpression)

Instantiate based on the provided SpEL expression to evaluate an RSocket endpoint route at runtime against a request message.

Parameters:

routeExpression - the SpEL expression to use.

Method Detail

setClientRSocketConnector

public void setClientRSocketConnector(ClientRSocketConnector clientRSocketConnector)

Configure a ClientRSocketConnector for client side requests based on the connection provided by the ClientRSocketConnector.getRSocketRequester(). In case of server side, an RSocketRequester must be provided in the RSocketRequesterMethodArgumentResolver.RSOCKET_REQUESTER_HEADER header of request message.

Parameters:

clientRSocketConnector - the ClientRSocketConnector to use.

setCommand

public void setCommand(RSocketOutboundGateway.Command command)

Configure a RSocketOutboundGateway.Command for RSocket request type.

Parameters:

command - the RSocketOutboundGateway.Command to use.

setCommandExpression

public void setCommandExpression(Expression commandExpression)

Configure a SpEL expression to evaluate a RSocketOutboundGateway.Command for RSocket request type at runtime against a request message.

Parameters:

commandExpression - the SpEL expression to use.

setPublisherElementType

public void setPublisherElementType(Class<?> publisherElementType)

Configure a type for a request Publisher elements.

Parameters:

publisherElementType - the type of the request Publisher elements.

See Also:

RSocketRequester.RequestSpec#data(Publisher, Class)

setPublisherElementTypeExpression

public void setPublisherElementTypeExpression(Expression publisherElementTypeExpression)

Configure a SpEL expression to evaluate a request Publisher elements type at runtime against a request message.

Parameters:

publisherElementTypeExpression - the expression to evaluate a type for the request Publisher elements.

See Also:

RSocketRequester.RequestSpec#data

setExpectedResponseType

public void setExpectedResponseType(Class<?> expectedResponseType)

Specify the expected response type for the RSocket response.

Parameters:

expectedResponseType - The expected type.

See Also:

setExpectedResponseTypeExpression(Expression), RSocketRequester.ResponseSpec#retrieveMono, RSocketRequester.ResponseSpec#retrieveFlux

setExpectedResponseTypeExpression

public void setExpectedResponseTypeExpression(Expression expectedResponseTypeExpression)

Specify the Expression to determine the type for the RSocket response.

Parameters:

expectedResponseTypeExpression - The expected response type expression.

See Also:

RSocketRequester.ResponseSpec#retrieveMono, RSocketRequester.ResponseSpec#retrieveFlux

doInit

protected void doInit()

Overrides:

doInit in class AbstractReplyProducingMessageHandler

handleRequestMessage

protected Object handleRequestMessage(Message<?> requestMessage)

Description copied from class: AbstractReplyProducingMessageHandler

Subclasses must implement this method to handle the request Message. The return value may be a Message, a MessageBuilder, or any plain Object. The base class will handle the final creation of a reply Message from any of those starting points. If the return value is null, the Message flow will end here.

Specified by:

handleRequestMessage in class AbstractReplyProducingMessageHandler

Parameters:

requestMessage - The request message.

Returns:

The result of handling the message, or null.