org.springframework.integration.zeromq.outbound.ZeroMqMessageHandler

All Implemented Interfaces:: Aware, BeanFactoryAware, BeanNameAware, DisposableBean, InitializingBean, ApplicationContextAware, Lifecycle, Ordered, ComponentSourceAware, 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 can bind or connect the socket.

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)

Create an instance based on the provided ZContext.

ZeroMqMessageHandler(org.zeromq.ZContext context, int port)

Create an instance based on the provided ZContext and binding port.

ZeroMqMessageHandler(org.zeromq.ZContext context, int port, org.zeromq.SocketType socketType)

Create an instance based on the provided ZContext, binding port and SocketType.

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.

ZeroMqMessageHandler(org.zeromq.ZContext context, org.zeromq.SocketType socketType)

Create an instance based on the provided ZContext and SocketType.
Method Summary

Modifier and Type

Method

Description

void

destroy()

int

getBoundPort()

Return the port a socket is bound or 0 if this message producer has not been started yet or the socket is connected - not bound.

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, getComponentDescription, getComponentName, getComponentSource, getConversionService, getExpression, getIntegrationProperties, getMessageBuilderFactory, getTaskScheduler, isInitialized, setApplicationContext, setBeanFactory, setBeanName, setChannelResolver, setComponentDescription, setComponentName, setComponentSource, 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)
  
  Create an instance based on the provided ZContext.
  
  Parameters:
  
  context - the ZContext to use for creating sockets.
  
  Since:
  
  6.4
- ZeroMqMessageHandler
  
  public ZeroMqMessageHandler(org.zeromq.ZContext context, org.zeromq.SocketType socketType)
  
  Create an instance based on the provided ZContext and SocketType.
  
  Parameters:
  
  context - the ZContext to use for creating sockets.
  
  socketType - the SocketType to use; only SocketType.PAIR, SocketType.PUB and SocketType.PUSH are supported.
- 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, int port)
  
  Create an instance based on the provided ZContext and binding port.
  
  Parameters:
  
  context - the ZContext to use for creating sockets.
  
  port - the port to bind ZeroMq socket to over TCP.
  
  Since:
  
  6.4
- 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, int port, org.zeromq.SocketType socketType)
  
  Create an instance based on the provided ZContext, binding port and SocketType.
  
  Parameters:
  
  context - the ZContext to use for creating sockets.
  
  port - the port to bind ZeroMq socket to over TCP.
  
  socketType - the SocketType to use; only SocketType.PAIR, SocketType.PUB and SocketType.PUSH are supported.
  
  Since:
  
  6.4
- 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
- getBoundPort
  
  public int getBoundPort()
  
  Return the port a socket is bound or 0 if this message producer has not been started yet or the socket is connected - not bound.
  
  Returns:
  
  the port for a socket or 0.
  
  Since:
  
  6.4
- 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

Class ZeroMqMessageHandler

Nested Class Summary

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

Field Summary

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

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

Fields inherited from interface org.springframework.core.Ordered

Constructor Summary

Method Summary

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

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

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

Methods inherited from class java.lang.Object

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

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

Constructor Details

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

ZeroMqMessageHandler

Method Details

setMessageMapper

setMessageConverter

setSocketConfigurer

setTopic

setTopicExpression

wrapTopic

getBoundPort

getComponentType

onInit

start

stop

isRunning

handleMessageInternal

destroy