org.springframework.integration.amqp.outbound

Class AbstractAmqpOutboundEndpoint

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.MessageHandlerSupport

org.springframework.integration.handler.AbstractMessageHandler

org.springframework.integration.handler.AbstractMessageProducingHandler

org.springframework.integration.handler.AbstractReplyProducingMessageHandler

org.springframework.integration.amqp.outbound.AbstractAmqpOutboundEndpoint

All Implemented Interfaces:

org.reactivestreams.Subscriber<Message<?>>, Aware, BeanClassLoaderAware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, Lifecycle, Ordered, ExpressionCapable, Orderable, MessageProducer, HeaderPropagationAware, IntegrationPattern, NamedComponent, IntegrationManagement, ManageableLifecycle, TrackableComponent, MessageHandler, reactor.core.CoreSubscriber<Message<?>>

Direct Known Subclasses:

AmqpOutboundEndpoint, AsyncAmqpOutboundGateway

public abstract class AbstractAmqpOutboundEndpoint
extends AbstractReplyProducingMessageHandler
implements ManageableLifecycle

Since:

4.3

Author:

Gary Russell, Artem Bilan

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	AbstractAmqpOutboundEndpoint.CorrelationDataWrapper

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.core.Ordered

HIGHEST_PRECEDENCE, LOWEST_PRECEDENCE

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

METER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME

Constructor Summary

Constructors

Constructor and Description
AbstractAmqpOutboundEndpoint()

Method Summary

Modifier and Type	Method and Description
protected void	addDelayProperty(Message<?> message, org.springframework.amqp.core.Message amqpMessage)
protected AbstractIntegrationMessageBuilder<?>	buildReply(org.springframework.amqp.support.converter.MessageConverter converter, org.springframework.amqp.core.Message amqpReplyMessage)
protected Message<?>	buildReturnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String returnedRoutingKey, org.springframework.amqp.support.converter.MessageConverter converter) Deprecated. since 5.4 in favor of buildReturnedMessage(ReturnedMessage, MessageConverter)
protected Message<?>	buildReturnedMessage(org.springframework.amqp.core.ReturnedMessage returnedMessage, org.springframework.amqp.support.converter.MessageConverter converter)
protected void	doInit()
protected void	doStart()
protected void	doStop()
protected void	endpointInit() Subclasses can override to perform any additional initialization.
protected org.springframework.amqp.rabbit.connection.CorrelationData	generateCorrelationData(Message<?> requestMessage)
protected String	generateExchangeName(Message<?> requestMessage)
protected String	generateRoutingKey(Message<?> requestMessage)
protected MessageChannel	getConfirmAckChannel()
protected Expression	getConfirmCorrelationExpression()
protected MessageChannel	getConfirmNackChannel()
protected java.time.Duration	getConfirmTimeout()
protected ExpressionEvaluatingMessageProcessor<Object>	getCorrelationDataGenerator()
protected org.springframework.amqp.core.MessageDeliveryMode	getDefaultDeliveryMode()
protected String	getExchangeName()
protected Expression	getExchangeNameExpression()
protected ExpressionEvaluatingMessageProcessor<String>	getExchangeNameGenerator()
AmqpHeaderMapper	getHeaderMapper()
protected abstract org.springframework.amqp.rabbit.core.RabbitTemplate	getRabbitTemplate()
protected MessageChannel	getReturnChannel()
protected String	getRoutingKey()
protected Expression	getRoutingKeyExpression()
protected ExpressionEvaluatingMessageProcessor<String>	getRoutingKeyGenerator()
protected void	handleConfirm(org.springframework.amqp.rabbit.connection.CorrelationData correlationData, boolean ack, String cause)
protected boolean	isHeadersMappedLast()
protected boolean	isLazyConnect()
boolean	isRunning()
void	setConfirmAckChannel(MessageChannel ackChannel) Set the channel to which acks are send (publisher confirms).
void	setConfirmAckChannelName(String ackChannelName) Set the channel name to which acks are send (publisher confirms).
void	setConfirmCorrelationExpression(Expression confirmCorrelationExpression) Set a SpEL expression to evaluate confirm correlation at runtime.
void	setConfirmCorrelationExpressionString(String confirmCorrelationExpression) Set a SpEL expression to evaluate confirm correlation at runtime.
void	setConfirmNackChannel(MessageChannel nackChannel) Set the channel to which nacks are send (publisher confirms).
void	setConfirmNackChannelName(String nackChannelName) Set the channel name to which nacks are send (publisher confirms).
void	setConfirmTimeout(long confirmTimeout) Set a timeout after which a nack will be synthesized if no publisher confirm has been received within that time.
protected void	setConnectionFactory(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory)
void	setDefaultDeliveryMode(org.springframework.amqp.core.MessageDeliveryMode defaultDeliveryMode) Set the default delivery mode.
void	setDelay(int delay) Set the value to set in the x-delay header when using the RabbitMQ delayed message exchange plugin.
void	setDelayExpression(Expression delayExpression) Set the SpEL expression to calculate the x-delay header when using the RabbitMQ delayed message exchange plugin.
void	setDelayExpressionString(String delayExpression) Set the SpEL expression to calculate the x-delay header when using the RabbitMQ delayed message exchange plugin.
void	setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy) Set the error message strategy to use for returned (or negatively confirmed) messages.
void	setExchangeName(String exchangeName) Configure an AMQP exchange name for sending messages.
void	setExchangeNameExpression(Expression exchangeNameExpression) Configure a SpEL expression to evaluate an exchange name at runtime.
void	setExchangeNameExpressionString(String exchangeNameExpression)
void	setHeaderMapper(AmqpHeaderMapper headerMapper) Set a custom AmqpHeaderMapper for mapping request and reply headers.
void	setHeadersMappedLast(boolean headersMappedLast) When mapping headers for the outbound message, determine whether the headers are mapped before the message is converted, or afterwards.
void	setLazyConnect(boolean lazyConnect) Set to false to attempt to connect during endpoint start; default true, meaning the connection will be attempted to be established on the arrival of the first message.
void	setReturnChannel(MessageChannel returnChannel) Set the channel to which returned messages are sent.
void	setRoutingKey(String routingKey) Configure an AMQP routing key for sending messages.
void	setRoutingKeyExpression(Expression routingKeyExpression)
void	setRoutingKeyExpressionString(String routingKeyExpression)
void	start()
void	stop()

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

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

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

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

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

handleMessage, onComplete, onError, onNext, onSubscribe

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

buildSendTimer, destroy, getComponentType, getManagedName, getManagedType, getMetricsCaptor, getOrder, getOverrides, isLoggingEnabled, registerMetricsCaptor, sendTimer, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, shouldTrack

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

afterPropertiesSet, extractTypeIfPossible, generateId, getApplicationContext, getApplicationContextId, getBeanDescription, 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 reactor.core.CoreSubscriber

currentContext

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

getThisAs

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

getBeanName, getComponentName

Constructor Detail

AbstractAmqpOutboundEndpoint

public AbstractAmqpOutboundEndpoint()

Method Detail

setHeaderMapper

public void setHeaderMapper(AmqpHeaderMapper headerMapper)

Set a custom AmqpHeaderMapper for mapping request and reply headers. Defaults to DefaultAmqpHeaderMapper.outboundMapper().

Parameters:

headerMapper - the AmqpHeaderMapper to use.

setHeadersMappedLast

public void setHeadersMappedLast(boolean headersMappedLast)

When mapping headers for the outbound message, determine whether the headers are mapped before the message is converted, or afterwards. This only affects headers that might be added by the message converter. When false, the converter's headers win; when true, any headers added by the converter will be overridden (if the source message has a header that maps to those headers). You might wish to set this to true, for example, when using a SimpleMessageConverter with a String payload that contains json; the converter will set the content type to text/plain which can be overridden to application/json by setting the AmqpHeaders.CONTENT_TYPE message header. Default: false.

Parameters:

headersMappedLast - true if headers are mapped after conversion.

Since:

5.0

setExchangeName

public void setExchangeName(String exchangeName)

Configure an AMQP exchange name for sending messages.

Parameters:

exchangeName - the exchange name for sending messages.

setExchangeNameExpression

public void setExchangeNameExpression(Expression exchangeNameExpression)

Configure a SpEL expression to evaluate an exchange name at runtime.

Parameters:

exchangeNameExpression - the expression to use.

Since:

4.3

setExchangeNameExpressionString

public void setExchangeNameExpressionString(String exchangeNameExpression)

Parameters:

exchangeNameExpression - the String in SpEL syntax.

Since:

4.3

setRoutingKey

public void setRoutingKey(String routingKey)

Configure an AMQP routing key for sending messages.

Parameters:

routingKey - the routing key to use

setRoutingKeyExpression

public void setRoutingKeyExpression(Expression routingKeyExpression)

Parameters:

routingKeyExpression - the expression to use.

Since:

4.3

setRoutingKeyExpressionString

public void setRoutingKeyExpressionString(String routingKeyExpression)

Parameters:

routingKeyExpression - the String in SpEL syntax.

Since:

4.3

setConfirmCorrelationExpression

public void setConfirmCorrelationExpression(Expression confirmCorrelationExpression)

Set a SpEL expression to evaluate confirm correlation at runtime.

Parameters:

confirmCorrelationExpression - the expression to use.

Since:

4.3

setConfirmCorrelationExpressionString

public void setConfirmCorrelationExpressionString(String confirmCorrelationExpression)

Set a SpEL expression to evaluate confirm correlation at runtime.

Parameters:

confirmCorrelationExpression - the String in SpEL syntax.

Since:

4.3

setConfirmAckChannel

public void setConfirmAckChannel(MessageChannel ackChannel)

Set the channel to which acks are send (publisher confirms).

Parameters:

ackChannel - the channel.

setConfirmAckChannelName

public void setConfirmAckChannelName(String ackChannelName)

Set the channel name to which acks are send (publisher confirms).

Parameters:

ackChannelName - the channel name.

Since:

4.3.12

setConfirmNackChannel

public void setConfirmNackChannel(MessageChannel nackChannel)

Set the channel to which nacks are send (publisher confirms).

Parameters:

nackChannel - the channel.

setConfirmNackChannelName

public void setConfirmNackChannelName(String nackChannelName)

Set the channel name to which nacks are send (publisher confirms).

Parameters:

nackChannelName - the channel name.

Since:

4.3.12

setReturnChannel

public void setReturnChannel(MessageChannel returnChannel)

Set the channel to which returned messages are sent.

Parameters:

returnChannel - the channel.

setDefaultDeliveryMode

public void setDefaultDeliveryMode(org.springframework.amqp.core.MessageDeliveryMode defaultDeliveryMode)

Set the default delivery mode.

Parameters:

defaultDeliveryMode - the delivery mode.

setLazyConnect

public void setLazyConnect(boolean lazyConnect)

Set to false to attempt to connect during endpoint start; default true, meaning the connection will be attempted to be established on the arrival of the first message.

Parameters:

lazyConnect - the lazyConnect to set

Since:

4.1

setDelay

public void setDelay(int delay)

Set the value to set in the x-delay header when using the RabbitMQ delayed message exchange plugin. By default, the AmqpHeaders.DELAY header (if present) is mapped; setting the delay here overrides that value.

Parameters:

delay - the delay.

Since:

4.3.5

setDelayExpression

public void setDelayExpression(Expression delayExpression)

Set the SpEL expression to calculate the x-delay header when using the RabbitMQ delayed message exchange plugin. By default, the AmqpHeaders.DELAY header (if present) is mapped; setting the expression here overrides that value.

Parameters:

delayExpression - the expression.

Since:

4.3.5

setDelayExpressionString

public void setDelayExpressionString(@Nullable
                                     String delayExpression)

Set the SpEL expression to calculate the x-delay header when using the RabbitMQ delayed message exchange plugin. By default, the AmqpHeaders.DELAY header (if present) is mapped; setting the expression here overrides that value.

Parameters:

delayExpression - the expression.

Since:

4.3.5

setErrorMessageStrategy

public void setErrorMessageStrategy(ErrorMessageStrategy errorMessageStrategy)

Set the error message strategy to use for returned (or negatively confirmed) messages.

Parameters:

errorMessageStrategy - the strategy.

Since:

4.3.12

setConfirmTimeout

public void setConfirmTimeout(long confirmTimeout)

Set a timeout after which a nack will be synthesized if no publisher confirm has been received within that time. Missing confirms will be checked every 50% of this value so the synthesized nack will be sent between 1x and 1.5x this timeout.

Parameters:

confirmTimeout - the approximate timeout.

Since:

5.2

See Also:

setConfirmNackChannel(MessageChannel)

setConnectionFactory

protected final void setConnectionFactory(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory)

getExchangeName

protected String getExchangeName()

getRoutingKey

protected String getRoutingKey()

getExchangeNameExpression

protected Expression getExchangeNameExpression()

getRoutingKeyExpression

protected Expression getRoutingKeyExpression()

getRoutingKeyGenerator

protected ExpressionEvaluatingMessageProcessor<String> getRoutingKeyGenerator()

getExchangeNameGenerator

protected ExpressionEvaluatingMessageProcessor<String> getExchangeNameGenerator()

getHeaderMapper

public AmqpHeaderMapper getHeaderMapper()

getConfirmCorrelationExpression

protected Expression getConfirmCorrelationExpression()

getCorrelationDataGenerator

protected ExpressionEvaluatingMessageProcessor<Object> getCorrelationDataGenerator()

getConfirmAckChannel

protected MessageChannel getConfirmAckChannel()

getConfirmNackChannel

protected MessageChannel getConfirmNackChannel()

getReturnChannel

protected MessageChannel getReturnChannel()

getDefaultDeliveryMode

protected org.springframework.amqp.core.MessageDeliveryMode getDefaultDeliveryMode()

isLazyConnect

protected boolean isLazyConnect()

isHeadersMappedLast

protected boolean isHeadersMappedLast()

getConfirmTimeout

@Nullable
protected java.time.Duration getConfirmTimeout()

doInit

protected final void doInit()

Overrides:

doInit in class AbstractReplyProducingMessageHandler

endpointInit

protected void endpointInit()

Subclasses can override to perform any additional initialization. Called from afterPropertiesSet().

start

public void start()

Specified by:

start in interface Lifecycle

Specified by:

start in interface ManageableLifecycle

getRabbitTemplate

@Nullable
protected abstract org.springframework.amqp.rabbit.core.RabbitTemplate getRabbitTemplate()

stop

public void stop()

Specified by:

stop in interface Lifecycle

Specified by:

stop in interface ManageableLifecycle

doStart

protected void doStart()

doStop

protected void doStop()

isRunning

public boolean isRunning()

Specified by:

isRunning in interface Lifecycle

Specified by:

isRunning in interface ManageableLifecycle

generateCorrelationData

protected org.springframework.amqp.rabbit.connection.CorrelationData generateCorrelationData(Message<?> requestMessage)

generateExchangeName

protected String generateExchangeName(Message<?> requestMessage)

generateRoutingKey

protected String generateRoutingKey(Message<?> requestMessage)

addDelayProperty

protected void addDelayProperty(Message<?> message,
                                org.springframework.amqp.core.Message amqpMessage)

buildReply

protected AbstractIntegrationMessageBuilder<?> buildReply(org.springframework.amqp.support.converter.MessageConverter converter,
                                                          org.springframework.amqp.core.Message amqpReplyMessage)

buildReturnedMessage

@Deprecated
protected Message<?> buildReturnedMessage(org.springframework.amqp.core.Message message,
                                                      int replyCode,
                                                      String replyText,
                                                      String exchange,
                                                      String returnedRoutingKey,
                                                      org.springframework.amqp.support.converter.MessageConverter converter)

Deprecated. since 5.4 in favor of buildReturnedMessage(ReturnedMessage, MessageConverter)

Build Spring message object based on the provided returned AMQP message info.

Parameters:

message - the returned AMQP message

replyCode - the returned message reason code

replyText - the returned message reason text

exchange - the exchange the message returned from

returnedRoutingKey - the routing key for returned message

converter - the converter to deserialize body of the returned AMQP message

Returns:

the Spring message which represents a returned AMQP message

buildReturnedMessage

protected Message<?> buildReturnedMessage(org.springframework.amqp.core.ReturnedMessage returnedMessage,
                                          org.springframework.amqp.support.converter.MessageConverter converter)

handleConfirm

protected void handleConfirm(org.springframework.amqp.rabbit.connection.CorrelationData correlationData,
                             boolean ack,
                             String cause)