public class ZeroMqMessageHandler extends AbstractReactiveMessageHandler
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.
IntegrationManagement.ManagementOverridesEXPRESSION_PARSER, loggerHIGHEST_PRECEDENCE, LOWEST_PRECEDENCEMETER_PREFIX, RECEIVE_COUNTER_NAME, SEND_TIMER_NAME| Constructor and 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. |
| Modifier and Type | Method and Description |
|---|---|
void |
destroy() |
String |
getComponentType()
Subclasses may implement this method to provide component type information.
|
protected reactor.core.publisher.Mono<Void> |
handleMessageInternal(Message<?> message) |
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(java.util.function.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. |
handleMessagebuildSendTimer, getIntegrationPatternType, getManagedName, getManagedType, getMetricsCaptor, getOrder, getOverrides, isLoggingEnabled, registerMetricsCaptor, sendTimer, setLoggingEnabled, setManagedName, setManagedType, setOrder, setShouldTrack, shouldTrackafterPropertiesSet, 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, toStringclone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, waitgetThisAsgetBeanName, getComponentNamepublic ZeroMqMessageHandler(org.zeromq.ZContext context,
String connectUrl)
ZContext and connection string.context - the ZContext to use for creating sockets.connectUrl - the URL to connect the socket to.public ZeroMqMessageHandler(org.zeromq.ZContext context,
String connectUrl,
org.zeromq.SocketType socketType)
ZContext, connection string and SocketType.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.public void setMessageMapper(OutboundMessageMapper<byte[]> messageMapper)
OutboundMessageMapper to convert a request message into byte[]
for sending into ZeroMq socket.
Ignored when Message.getPayload() is an instance of ZMsg.messageMapper - the OutboundMessageMapper to use.public void setMessageConverter(MessageConverter messageConverter)
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.messageConverter - the MessageConverter to use.public void setSocketConfigurer(java.util.function.Consumer<org.zeromq.ZMQ.Socket> socketConfigurer)
Consumer to configure a socket with arbitrary options, like security.socketConfigurer - the configurer for socket options.public void setTopic(String topic)
SocketType.PUB socket is going to use for distributing messages into the
subscriptions. It is ignored for all other SocketTypes supported.topic - the topic to use.public void setTopicExpression(Expression topicExpression)
SocketType.PUB
is going to use for distributing messages into the
subscriptions.It is ignored for all other SocketTypes supported.topicExpression - the expression to evaluate topic for publishing.public String getComponentType()
IntegrationObjectSupportgetComponentType in interface NamedComponentgetComponentType in class MessageHandlerSupportprotected void onInit()
IntegrationObjectSupportonInit in class IntegrationObjectSupportprotected reactor.core.publisher.Mono<Void> handleMessageInternal(Message<?> message)
handleMessageInternal in class AbstractReactiveMessageHandlerpublic void destroy()
destroy in interface DisposableBeandestroy in interface IntegrationManagementdestroy in class MessageHandlerSupport