Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-9197: Optional ZeroMQ topic wrapping

Browse Source 
Fixes: #9197

* Update `ZeroMqMessageHandler` for `wrapTopic` option
* Add author, fix code style, add same logic also for `ZeroMqMessageProducer`
* Add `wrapTopic` function also in DSL specs
* Fix wrap topic test: duplicate socket address caused binding exception
* Rewrite the `MessageProducer.wrapTopic()` test
* Call `stop()` instead of `destroy()` method

(cherry picked from commit 80c8a61f33)
This commit is contained in:
Alessio Matricardi
2024-06-10 16:56:07 +02:00
committed by Spring Builds
parent 460a4b670b
commit ac29b2673e
7 changed files with 156 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/reference/antora/modules/ROOT/pages/zeromq.adoc
View File

@@ -119,6 +119,9 @@ If the `receiveRaw` option is set to `true`, a `ZMsg`, consumed from the socket,
Otherwise, an `InboundMessageMapper` is used to convert the consumed data into a `Message`.
If the received `ZMsg` is multi-frame, the first frame is treated as the `ZeroMqHeaders.TOPIC` header this ZeroMQ message was published to.
If the `unwrapTopic` option is set to `false`, the incoming message is considered to consist of two frames: the topic and the ZeroMQ message.
Otherwise, by default, the `ZMsg` is considered to consist of three frames: the first one containing the topic, the last frame containing the message, with an empty frame in the middle.
With `SocketType.SUB`, the `ZeroMqMessageProducer` uses the provided `topics` option for subscriptions; defaults to subscribe to all.
Subscriptions can be adjusted at runtime using `subscribeToTopics()` and `unsubscribeFromTopics()` `@ManagedOperation` s.
@@ -146,6 +149,10 @@ Only `SocketType.PAIR`, `SocketType.PUSH` and `SocketType.PUB` are supported.
The `ZeroMqMessageHandler` only supports connecting the ZeroMQ socket; binding is not supported.
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 (`SocketType.SUB`) must receive the topic frame first before parsing the actual data.
If the `wrapTopic` option is set to `false`, the ZeroMQ message frame is sent after the injected topic, if present.
By default, an additional empty frame is sent between the topic and the message.
When the payload of the request message is a `ZMsg`, no conversion or topic extraction is performed: the `ZMsg` is sent into a socket as is and it is not destroyed for possible further reuse.
Otherwise, an `OutboundMessageMapper<byte[]>` is used to convert a request message (or just its payload) into a ZeroMQ frame to publish.
By default, a `ConvertingBytesMessageMapper` is used supplied with a `ConfigurableCompositeMessageConverter`.