Package org.springframework.integration.zeromq.outbound

Class ZeroMqMessageHandler

java.lang.Object

org.springframework.integration.context.IntegrationObjectSupport

org.springframework.integration.handler.MessageHandlerSupport

org.springframework.integration.handler.AbstractReactiveMessageHandler

org.springframework.integration.zeromq.outbound.ZeroMqMessageHandler

All Implemented Interfaces:

Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, Lifecycle, Ordered, ExpressionCapable, Orderable, IntegrationPattern, NamedComponent, IntegrationManagement, ManageableLifecycle, TrackableComponent, ReactiveMessageHandler

public class ZeroMqMessageHandler
extends AbstractReactiveMessageHandler
implements ManageableLifecycle

The AbstractReactiveMessageHandler implementation for publishing messages over ZeroMq socket. Only SocketType.PAIR, SocketType.PUB and SocketType.PUSH are supported. This component is only connecting (no Binding) to another side, e.g. ZeroMq proxy.

When the SocketType.PUB is used, the topicExpression is evaluated against a request message to inject a topic frame into a ZeroMq message if it is not null. The subscriber side must receive the topic frame first before parsing the actual data.

When the payload of the request message is a ZMsg, no any conversion and topic extraction happen: the ZMsg is sent into a socket as is and it is not destroyed for possible further reusing.

Since:

5.4

Author:

Artem Bilan, Alessio Matricardi

Nested Class Summary

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

IntegrationManagement.ManagementOverrides

Field Summary

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	Description
ZeroMqMessageHandler(org.zeromq.ZContext context, String connectUrl)	Create an instance based on the provided ZContext and connection string.
ZeroMqMessageHandler(org.zeromq.ZContext context, String connectUrl, org.zeromq.SocketType socketType)	Create an instance based on the provided ZContext, connection string and SocketType.
ZeroMqMessageHandler(org.zeromq.ZContext context, Supplier<String> connectUrl)	Create an instance based on the provided ZContext and connection string supplier.
ZeroMqMessageHandler(org.zeromq.ZContext context, Supplier<String> connectUrl, org.zeromq.SocketType socketType)	Create an instance based on the provided ZContext, connection string supplier and SocketType.

Method Summary

Modifier and Type	Method	Description
void	destroy()
String	getComponentType()	Subclasses may implement this method to provide component type information.
protected reactor.core.publisher.Mono<Void>	handleMessageInternal(Message<?> message)
boolean	isRunning()
protected void	onInit()	Subclasses may implement this for initialization logic.
void	setMessageConverter(MessageConverter messageConverter)	Provide a MessageConverter (as an alternative to messageMapper) for converting a request message into byte[] for sending into ZeroMq socket.
void	setMessageMapper(OutboundMessageMapper<byte[]> messageMapper)	Provide an OutboundMessageMapper to convert a request message into byte[] for sending into ZeroMq socket.
void	setSocketConfigurer(Consumer<org.zeromq.ZMQ.Socket> socketConfigurer)	Provide a Consumer to configure a socket with arbitrary options, like security.
void	setTopic(String topic)	Specify a topic the SocketType.PUB socket is going to use for distributing messages into the subscriptions.
void	setTopicExpression(Expression topicExpression)	Specify a SpEL expression to evaluate a topic a SocketType.PUB is going to use for distributing messages into the subscriptions.It is ignored for all other SocketTypes supported.
void	start()
void	stop()
void	wrapTopic(boolean wrapTopic)	Specify if the topic that SocketType.PUB socket is going to use for distributing messages into the subscriptions must be wrapped with an additional empty frame.

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

handleMessage

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

buildSendTimer, getIntegrationPatternType, getManagedName, getManagedType, getMetricsCaptor, getObservationRegistry, getOrder, getOverrides, isLoggingEnabled, isObserved, registerMetricsCaptor, registerObservationRegistry, 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, 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.IntegrationManagement

getThisAs

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

getBeanName, getComponentName

Constructor Details

ZeroMqMessageHandler

public ZeroMqMessageHandler(org.zeromq.ZContext context,
 String connectUrl)

Create an instance based on the provided ZContext and connection string.

Parameters:

context - the ZContext to use for creating sockets.

connectUrl - the URL to connect the socket to.

ZeroMqMessageHandler

public ZeroMqMessageHandler(org.zeromq.ZContext context,
 Supplier<String> connectUrl)

Create an instance based on the provided ZContext and connection string supplier.

Parameters:

context - the ZContext to use for creating sockets.

connectUrl - the supplier for URL to connect the socket to.

Since:

5.5.9

ZeroMqMessageHandler

public ZeroMqMessageHandler(org.zeromq.ZContext context,
 String connectUrl,
 org.zeromq.SocketType socketType)

Create an instance based on the provided ZContext, connection string and SocketType.

Parameters:

context - the ZContext to use for creating sockets.

connectUrl - the URL to connect the socket to.

socketType - the SocketType to use; only SocketType.PAIR, SocketType.PUB and SocketType.PUSH are supported.

ZeroMqMessageHandler

public ZeroMqMessageHandler(org.zeromq.ZContext context,
 Supplier<String> connectUrl,
 org.zeromq.SocketType socketType)

Create an instance based on the provided ZContext, connection string supplier and SocketType.

Parameters:

context - the ZContext to use for creating sockets.

connectUrl - the supplier for URL to connect the socket to.

socketType - the SocketType to use; only SocketType.PAIR, SocketType.PUB and SocketType.PUSH are supported.

Since:

5.5.9

Method Details

setMessageMapper

public void setMessageMapper(OutboundMessageMapper<byte[]> messageMapper)

Provide an OutboundMessageMapper to convert a request message into byte[] for sending into ZeroMq socket. Ignored when Message.getPayload() is an instance of ZMsg.

Parameters:

messageMapper - the OutboundMessageMapper to use.

setMessageConverter

public void setMessageConverter(MessageConverter messageConverter)

Provide a MessageConverter (as an alternative to messageMapper) for converting a request message into byte[] for sending into ZeroMq socket. Ignored when Message.getPayload() is an instance of ZMsg.

Parameters:

messageConverter - the MessageConverter to use.

setSocketConfigurer

public void setSocketConfigurer(Consumer<org.zeromq.ZMQ.Socket> socketConfigurer)

Provide a Consumer to configure a socket with arbitrary options, like security.

Parameters:

socketConfigurer - the configurer for socket options.

setTopic

public void setTopic(String topic)

Specify a topic the SocketType.PUB socket is going to use for distributing messages into the subscriptions. It is ignored for all other SocketTypes supported.

Parameters:

topic - the topic to use.

setTopicExpression

public void setTopicExpression(Expression topicExpression)

Specify a SpEL expression to evaluate a topic a SocketType.PUB is going to use for distributing messages into the subscriptions.It is ignored for all other SocketTypes supported.

Parameters:

topicExpression - the expression to evaluate topic for publishing.

wrapTopic

public void wrapTopic(boolean wrapTopic)

Specify if the topic that SocketType.PUB socket is going to use for distributing messages into the subscriptions must be wrapped with an additional empty frame. It is ignored for all other SocketTypes supported. This attribute is set to true by default.

Parameters:

wrapTopic - true if the topic must be wrapped with an additional empty frame.

Since:

6.2.6

getComponentType

public String getComponentType()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this method to provide component type information.

Specified by:

getComponentType in interface NamedComponent

Overrides:

getComponentType in class MessageHandlerSupport

onInit

protected void onInit()

Description copied from class: IntegrationObjectSupport

Subclasses may implement this for initialization logic.

Overrides:

onInit in class IntegrationObjectSupport

start

public void start()

Specified by:

start in interface Lifecycle

Specified by:

start in interface ManageableLifecycle

stop

public void stop()

Specified by:

stop in interface Lifecycle

Specified by:

stop in interface ManageableLifecycle

isRunning

public boolean isRunning()

Specified by:

isRunning in interface Lifecycle

Specified by:

isRunning in interface ManageableLifecycle

handleMessageInternal

protected reactor.core.publisher.Mono<Void> handleMessageInternal(Message<?> message)

Specified by:

handleMessageInternal in class AbstractReactiveMessageHandler

destroy

public void destroy()

Specified by:

destroy in interface DisposableBean

Specified by:

destroy in interface IntegrationManagement

Overrides:

destroy in class MessageHandlerSupport