org.springframework.messaging.simp.stomp

Class StompBrokerRelayMessageHandler

java.lang.Object

org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler

org.springframework.messaging.simp.stomp.StompBrokerRelayMessageHandler

All Implemented Interfaces:

Aware, ApplicationEventPublisherAware, Lifecycle, Phased, SmartLifecycle, MessageHandler

public class StompBrokerRelayMessageHandler
extends AbstractBrokerMessageHandler

A MessageHandler that handles messages by forwarding them to a STOMP broker.

For each new CONNECT message, an independent TCP connection to the broker is opened and used exclusively for all messages from the client that originated the CONNECT message. Messages from the same client are identified through the session id message header. Reversely, when the STOMP broker sends messages back on the TCP connection, those messages are enriched with the session id of the client and sent back downstream through the MessageChannel provided to the constructor.

This class also automatically opens a default "system" TCP connection to the message broker that is used for sending messages that originate from the server application (as opposed to from a client). Such messages are are not associated with any client and therefore do not have a session id header. The "system" connection is effectively shared and cannot be used to receive messages. Several properties are provided to configure the "system" connection including:

• setSystemLogin(String)
• setSystemPasscode(String)
• setSystemHeartbeatSendInterval(long)
• setSystemHeartbeatReceiveInterval(long)

Since:

4.0

Author:

Rossen Stoyanchev, Andy Wilkinson

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
private class	StompBrokerRelayMessageHandler.Stats
private class	StompBrokerRelayMessageHandler.StompConnectionHandler
private static class	StompBrokerRelayMessageHandler.StompTcpClientFactory
private class	StompBrokerRelayMessageHandler.SystemStompConnectionHandler
private static class	StompBrokerRelayMessageHandler.VoidCallable

Field Summary

Fields

Modifier and Type	Field and Description
private java.lang.String	clientLogin
private java.lang.String	clientPasscode
private java.util.Map<java.lang.String,StompBrokerRelayMessageHandler.StompConnectionHandler>	connectionHandlers
private static byte[]	EMPTY_PAYLOAD
private static ListenableFutureTask<java.lang.Void>	EMPTY_TASK
private MessageHeaderInitializer	headerInitializer
private static Message<byte[]>	HEARTBEAT_MESSAGE
private static long	HEARTBEAT_MULTIPLIER
private java.lang.String	relayHost
private int	relayPort
private StompBrokerRelayMessageHandler.Stats	stats
static java.lang.String	SYSTEM_SESSION_ID
private long	systemHeartbeatReceiveInterval
private long	systemHeartbeatSendInterval
private java.lang.String	systemLogin
private java.lang.String	systemPasscode
private java.util.Map<java.lang.String,MessageHandler>	systemSubscriptions
private TcpOperations<byte[]>	tcpClient
private java.lang.String	virtualHost

Fields inherited from class org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler

logger

Constructor Summary

Constructors

Constructor and Description
StompBrokerRelayMessageHandler(SubscribableChannel inboundChannel, MessageChannel outboundChannel, SubscribableChannel brokerChannel, java.util.Collection<java.lang.String> destinationPrefixes) Create a StompBrokerRelayMessageHandler instance with the given message channels and destination prefixes.

Method Summary

Modifier and Type	Method and Description
java.lang.String	getClientLogin() Return the configured login to use for connections to the STOMP broker on behalf of connected clients.
java.lang.String	getClientPasscode() Return the configured passcode to use for connections to the STOMP broker on behalf of connected clients.
int	getConnectionCount() Return the current count of TCP connection to the broker.
MessageHeaderInitializer	getHeaderInitializer() Return the configured header initializer.
java.lang.String	getRelayHost() Return the STOMP message broker host.
int	getRelayPort() Return the STOMP message broker port.
java.lang.String	getStatsInfo() Return a String describing internal state and counters.
long	getSystemHeartbeatReceiveInterval() Return the interval, in milliseconds, at which the "system" connection expects to receive heartbeats from the STOMP broker.
long	getSystemHeartbeatSendInterval() Return the interval, in milliseconds, at which the "system" connection will send heartbeats to the STOMP broker.
java.lang.String	getSystemLogin() Return the login used for the shared "system" connection to the STOMP broker.
java.lang.String	getSystemPasscode() Return the passcode used for the shared "system" connection to the STOMP broker.
java.util.Map<java.lang.String,MessageHandler>	getSystemSubscriptions() Return the configured map with subscriptions on the "system" connection.
TcpOperations<byte[]>	getTcpClient() Get the configured TCP client.
java.lang.String	getVirtualHost() Return the configured virtual host value.
protected void	handleMessageInternal(Message<?> message)
void	setClientLogin(java.lang.String clientLogin) Set the login to use when creating connections to the STOMP broker on behalf of connected clients.
void	setClientPasscode(java.lang.String clientPasscode) Set the client passcode to use to create connections to the STOMP broker on behalf of connected clients.
void	setHeaderInitializer(MessageHeaderInitializer headerInitializer) Configure a MessageHeaderInitializer to apply to the headers of all messages created through the StompBrokerRelayMessageHandler that are sent to the client outbound message channel.
void	setRelayHost(java.lang.String relayHost) Set the STOMP message broker host.
void	setRelayPort(int relayPort) Set the STOMP message broker port.
void	setSystemHeartbeatReceiveInterval(long heartbeatReceiveInterval) Set the maximum interval, in milliseconds, at which the "system" connection expects, in the absence of any other data, to receive a heartbeat from the STOMP broker.
void	setSystemHeartbeatSendInterval(long systemHeartbeatSendInterval) Set the interval, in milliseconds, at which the "system" connection will, in the absence of any other data being sent, send a heartbeat to the STOMP broker.
void	setSystemLogin(java.lang.String systemLogin) Set the login for the shared "system" connection used to send messages to the STOMP broker from within the application, i.e.
void	setSystemPasscode(java.lang.String systemPasscode) Set the passcode for the shared "system" connection used to send messages to the STOMP broker from within the application, i.e.
void	setSystemSubscriptions(java.util.Map<java.lang.String,MessageHandler> subscriptions) Configure one more destinations to subscribe to on the shared "system" connection along with MessageHandler's to handle received messages.
void	setTcpClient(TcpOperations<byte[]> tcpClient) Configure a TCP client for managing TCP connections to the STOMP broker.
void	setVirtualHost(java.lang.String virtualHost) Set the value of the "host" header to use in STOMP CONNECT frames.
protected void	startInternal()
protected void	stopInternal()
java.lang.String	toString()

Methods inherited from class org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler

checkDestinationPrefix, getApplicationEventPublisher, getBrokerChannel, getClientInboundChannel, getClientOutboundChannel, getDestinationPrefixes, getPhase, handleMessage, isAutoStartup, isBrokerAvailable, isRunning, publishBrokerAvailableEvent, publishBrokerUnavailableEvent, setApplicationEventPublisher, setAutoStartup, start, stop, stop

Methods inherited from class java.lang.Object

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

Field Detail

SYSTEM_SESSION_ID

public static final java.lang.String SYSTEM_SESSION_ID

See Also:

Constant Field Values

EMPTY_PAYLOAD

private static final byte[] EMPTY_PAYLOAD

EMPTY_TASK

private static final ListenableFutureTask<java.lang.Void> EMPTY_TASK

HEARTBEAT_MULTIPLIER

private static final long HEARTBEAT_MULTIPLIER

See Also:

Constant Field Values

HEARTBEAT_MESSAGE

private static final Message<byte[]> HEARTBEAT_MESSAGE

relayHost

private java.lang.String relayHost

relayPort

private int relayPort

clientLogin

private java.lang.String clientLogin

clientPasscode

private java.lang.String clientPasscode

systemLogin

private java.lang.String systemLogin

systemPasscode

private java.lang.String systemPasscode

systemHeartbeatSendInterval

private long systemHeartbeatSendInterval

systemHeartbeatReceiveInterval

private long systemHeartbeatReceiveInterval

virtualHost

private java.lang.String virtualHost

systemSubscriptions

private final java.util.Map<java.lang.String,MessageHandler> systemSubscriptions

tcpClient

private TcpOperations<byte[]> tcpClient

headerInitializer

private MessageHeaderInitializer headerInitializer

connectionHandlers

private final java.util.Map<java.lang.String,StompBrokerRelayMessageHandler.StompConnectionHandler> connectionHandlers

stats

private final StompBrokerRelayMessageHandler.Stats stats

Constructor Detail

StompBrokerRelayMessageHandler

public StompBrokerRelayMessageHandler(SubscribableChannel inboundChannel,
                                      MessageChannel outboundChannel,
                                      SubscribableChannel brokerChannel,
                                      java.util.Collection<java.lang.String> destinationPrefixes)

Create a StompBrokerRelayMessageHandler instance with the given message channels and destination prefixes.

Parameters:

inboundChannel - the channel for receiving messages from clients (e.g. WebSocket clients)

outboundChannel - the channel for sending messages to clients (e.g. WebSocket clients)

brokerChannel - the channel for the application to send messages to the broker

destinationPrefixes - the broker supported destination prefixes; destinations that do not match the given prefix are ignored.

Method Detail

setRelayHost

public void setRelayHost(java.lang.String relayHost)

Set the STOMP message broker host.

getRelayHost

public java.lang.String getRelayHost()

Return the STOMP message broker host.

setRelayPort

public void setRelayPort(int relayPort)

Set the STOMP message broker port.

getRelayPort

public int getRelayPort()

Return the STOMP message broker port.

setSystemHeartbeatSendInterval

public void setSystemHeartbeatSendInterval(long systemHeartbeatSendInterval)

Set the interval, in milliseconds, at which the "system" connection will, in the absence of any other data being sent, send a heartbeat to the STOMP broker. A value of zero will prevent heartbeats from being sent to the broker.

The default value is 10000.

See class-level documentation for more information on the "system" connection.

getSystemHeartbeatSendInterval

public long getSystemHeartbeatSendInterval()

Return the interval, in milliseconds, at which the "system" connection will send heartbeats to the STOMP broker.

setSystemHeartbeatReceiveInterval

public void setSystemHeartbeatReceiveInterval(long heartbeatReceiveInterval)

Set the maximum interval, in milliseconds, at which the "system" connection expects, in the absence of any other data, to receive a heartbeat from the STOMP broker. A value of zero will configure the connection to expect not to receive heartbeats from the broker.

The default value is 10000.

See class-level documentation for more information on the "system" connection.

getSystemHeartbeatReceiveInterval

public long getSystemHeartbeatReceiveInterval()

Return the interval, in milliseconds, at which the "system" connection expects to receive heartbeats from the STOMP broker.

setClientLogin

public void setClientLogin(java.lang.String clientLogin)

Set the login to use when creating connections to the STOMP broker on behalf of connected clients.

By default this is set to "guest".

See Also:

setSystemLogin(String)

getClientLogin

public java.lang.String getClientLogin()

Return the configured login to use for connections to the STOMP broker on behalf of connected clients.

See Also:

getSystemLogin()

setClientPasscode

public void setClientPasscode(java.lang.String clientPasscode)

Set the client passcode to use to create connections to the STOMP broker on behalf of connected clients.

By default this is set to "guest".

See Also:

setSystemPasscode(java.lang.String)

getClientPasscode

public java.lang.String getClientPasscode()

Return the configured passcode to use for connections to the STOMP broker on behalf of connected clients.

See Also:

getSystemPasscode()

setSystemLogin

public void setSystemLogin(java.lang.String systemLogin)

Set the login for the shared "system" connection used to send messages to the STOMP broker from within the application, i.e. messages not associated with a specific client session (e.g. REST/HTTP request handling method).

By default this is set to "guest".

getSystemLogin

public java.lang.String getSystemLogin()

Return the login used for the shared "system" connection to the STOMP broker.

setSystemPasscode

public void setSystemPasscode(java.lang.String systemPasscode)

Set the passcode for the shared "system" connection used to send messages to the STOMP broker from within the application, i.e. messages not associated with a specific client session (e.g. REST/HTTP request handling method).

By default this is set to "guest".

getSystemPasscode

public java.lang.String getSystemPasscode()

Return the passcode used for the shared "system" connection to the STOMP broker.

setSystemSubscriptions

public void setSystemSubscriptions(java.util.Map<java.lang.String,MessageHandler> subscriptions)

Configure one more destinations to subscribe to on the shared "system" connection along with MessageHandler's to handle received messages.

This is for internal use in a multi-application server scenario where servers forward messages to each other (e.g. unresolved user destinations).

Parameters:

subscriptions - the destinations to subscribe to.

getSystemSubscriptions

public java.util.Map<java.lang.String,MessageHandler> getSystemSubscriptions()

Return the configured map with subscriptions on the "system" connection.

setVirtualHost

public void setVirtualHost(java.lang.String virtualHost)

Set the value of the "host" header to use in STOMP CONNECT frames. When this property is configured, a "host" header will be added to every STOMP frame sent to the STOMP broker. This may be useful for example in a cloud environment where the actual host to which the TCP connection is established is different from the host providing the cloud-based STOMP service.

By default this property is not set.

getVirtualHost

public java.lang.String getVirtualHost()

Return the configured virtual host value.

setTcpClient

public void setTcpClient(TcpOperations<byte[]> tcpClient)

Configure a TCP client for managing TCP connections to the STOMP broker. By default Reactor2TcpClient is used.

getTcpClient

public TcpOperations<byte[]> getTcpClient()

Get the configured TCP client. Never null unless not configured invoked and this method is invoked before the handler is started and hence a default implementation initialized.

getConnectionCount

public int getConnectionCount()

Return the current count of TCP connection to the broker.

setHeaderInitializer

public void setHeaderInitializer(MessageHeaderInitializer headerInitializer)

Configure a MessageHeaderInitializer to apply to the headers of all messages created through the StompBrokerRelayMessageHandler that are sent to the client outbound message channel.

By default this property is not set.

getHeaderInitializer

public MessageHeaderInitializer getHeaderInitializer()

Return the configured header initializer.

getStatsInfo

public java.lang.String getStatsInfo()

Return a String describing internal state and counters.

startInternal

protected void startInternal()

Overrides:

startInternal in class AbstractBrokerMessageHandler

stopInternal

protected void stopInternal()

Overrides:

stopInternal in class AbstractBrokerMessageHandler

handleMessageInternal

protected void handleMessageInternal(Message<?> message)

Specified by:

handleMessageInternal in class AbstractBrokerMessageHandler

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object