* Destroy proxies in the end of tests initiating removal for their internal executors
* Use `@Qualifier("subPubZeroMqProxy")` in the `ZeroMqDslTests.ContextConfiguration` to avoid
warning about deprecated `LocalVariableTableParameterNameDiscoverer`
Fixes https://github.com/spring-projects/spring-integration/issues/3686
* Add `src/idea` with respective editor config for IntelliJ IDEA.
Must be imported into an IDE
* Add `src/eclipse` with respective editor config for Eclipse/STS
* Reformat imports in source code according a new editor config
Fixes https://github.com/spring-projects/spring-integration/issues/3623
* `IntegrationFlow` refactoring
* Apply several code style improvements and good practices
* Code style: no empty lines for methods javadocs
* make deprecated implementation reuse actual one instead of the copy-paste approach
* add whats-new comments
* Fix whats-new page according to standards
* Remove usage of non-stable `org.springframework.util.SocketUtils`
* Replace it with `0` for those tests where it is possible to select OS port
* Remove the mentioning of the `SocketUtils` from the `testing.adoc`
* Use `TransportConstants.DEFAULT_STOMP_PORT` for `StompServerIntegrationTests`.
We may disable this test in the future for CI if it is not going to be stable
* Introduce `Supplier<String> connectUrl` variants for `ZeroMqMessageHandler`
to let it defer connection evaluation until subscription to the socket `Mono`
in the `ZeroMqMessageHandler`.
* Move connection logic in the `ZeroMqMessageHandler` to `Lifecycle.start()`
Related to https://github.com/spring-projects/spring-framework/issues/28054
**Cherry-pick to `5.4.x`**
Use `receiveTimeOut = 0` on SUB socket and rely
on the retries from the `await().atMost()`.
Also resubscribe in the retry callback before publishing the next attempt
Turns out PUB socket doesn't care if there are subscribers to it or not.
The sent message may be just lost in between.
* Resend message in the test until it is received by subscriber
* Use `await().untilAsserted()` to iterate the logic at most 10 seconds
SO: https://stackoverflow.com/questions/67214907/zeromq-with-spring-spring-integration-zeromq
The `Mono` is created in several places in ZeroMQ components from their constructors.
That leads to the reactive stream to be configured just after ctor, which will ignore
any changes to the options which are used from that `Mono` definition.
For example this code `.doOnNext(this.sendSocketConfigurer)` is done once during
reactive stream definition.
* Fix all the ZeroMQ components to defer usage of the options which could be changed
after ctor initialization
* Cover affected option changes in the tests
**Cherry-pick to `5.4.x`**
* INT-3045: Add in & out ZeroMq channel adapters
JIRA: https://jira.spring.io/browse/INT-3045
* Add `ZeroMqMessageHandler` to produce messages into one-way ZeroMq sockets
* Add `ZeroMqMessageProducer` to consumer messages from one-way ZeroMq sockets
* Add `ConvertingBytesMessageMapper` impl for the `BytesMessageMapper` to
delegate an actual conversion into the provided `MessageConverter`
* Add `ZeroMqHeaders` for message headers constants representing ZeroMq message attributes
* Fix `ZeroMqChannel` for the proper deferred `zeroMqProxy` evaluation
* Add more JavaDocs
* Fix `ZeroMqChannelTests.testPubSubBind()` to be sure that really all the
subscribed channels get the same message from the `PUB` socket
* * Fix typo in the `ConvertingBytesMessageMapper`
* Add `this` for `doOnError()` in the `ZeroMqChannel` & `ZeroMqMessageProducer`
* Change the bind logic in the `ZeroMqMessageProducer` to `port` and let it to
bind to random port.
The actual port is available later via `getBoundPort()`
* Introduce a `ZeroMqMessageProducer.receiveRaw()` to let received `ZMsg` to
be produce as a `payload`
* Add a logic into `ZeroMqMessageHandler` to treat `ZMsg` in the payload of
request message as is without any conversion
* Fix race condition in the `ZeroMqMessageProducer` to destroy `consumerScheduler`
when the main `Flux` is complete
* * Add Java DSL for ZeroMq components
* Extract `ReactiveMessageHandlerSpec` for `ReactiveMessageHandler` impls
* Add debug message into `EmbeddedJsonHeadersMessageMapper` when cannot `decodeNativeFormat()`
* Make `ReactiveMongoDbMessageHandlerSpec` extending `ReactiveMessageHandlerSpec`
* Make `ZeroMqProxy` `autoStartup` by default
* Add `ZeroMqDslTests` to cover all the Java DSL for ZeroMq
* Introduce a `MimeTypeSerializer` to serialize a `MimeType` into JSON as a plain string;
use it as extra serializer in the `JacksonJsonUtils.messagingAwareMapper()`
* Fix typo for the `AllowListTypeResolverBuilder` inner class
* * Add some docs
* Fix Checkstyle violations
* * More docs
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* INT-3045: Add `ZeroMqChannel` support
JIRA: https://jira.spring.io/browse/INT-3045
Provide a `SubscribableChannel` implementation for ZeroMQ
The general idea is to let to have a distributed channel implementation
where every client can connect to a single server backed by the channel.
The logic in the channel is fully transparent for end-user and there is just
enough to send message to it and subscribe for receiving on the other side.
If PUB/SUB model is used, all the subscribes (even over the network) going to
receive the same published message.
In case of PUSH/PULL only one subscriber in the whole cluster is going to get
the published message
* Use Reactor for better threading control
* JeroMQ is not interruptible-friendly: use control sockets to stop proxy loop
* Name Reactor's schedulers to avoid daemon threads
* * Use try-catch-with-resource to close sockets automatically
* Fix Checkstyle violations
* Use `Mono.handle()` to receive data from the socket
* * Optimize local for just a couple of PAIR sockets
* Implement TCP binding
* Add PUB/SUB tests
* * Fix subscriber scheduler name
* Optimize socket create logic
* Add PUSH/PULL over TCP test
* * Fix subscriber scheduler name
* Optimize socket create logic
* Add PUSH/PULL over TCP test
* Implement PUB/SUB over TCP
* * Introduce `ZeroMqProxy` - Spring-friendly component to configure and manage ZeroMq proxy
* Use this `ZeroMqProxy` logic as an external component for `ZeroMqChannel` testing
* * Fix Checkstyle
* Apply docs polishing
* Expose a capture socket on the proxy
* Implement `DisposableBean` in the `ZeroMqProxy` to destroy an internal executor service
* Add JavaDocs to `ZeroMqChannel`
* Add one more `ZeroMqChannel` to TCP test to be sure that proxy distribution works well
* * Add `hamcrest-core` dependency for Awatility
* * Add more JavaDocs to `ZeroMqProxy` and `ZeroMqChannel`
* Expose `ZeroMqChannel.setZeroMqProxy()` option for easier
configuration within the same application context
* Make `ZeroMqChannel` sockets configuration and connection
dependant on provided `ZeroMqProxy` (if any)
* Add `Consumer<ZMQ.Socket>` configuration callbacks to the `ZeroMqChannel`
* Expose `ZeroMqChannel.consumeDelay` option
* * Add docs for ZeroMQ
* Some additions into a `reactive-streams.adoc`
* Fix typo in the `xmpp.adoc`
* * Add `optional` `jackson-databind` since `ZeroMqChannel` uses it by default
* More words into docs
* * Fix language in docs according review
* Fix language in docs according review
Co-authored-by: Gary Russell <grussell@vmware.com>
* Apply suggestions from code review
Co-authored-by: Oliver <oli-ver@users.noreply.github.com>
* * Fix threading using a `publishOn()` for specific scheduler after `cache()`
* * Remove unused import
* * Change proxy port check from static `Mono.just()` to `Mono.fromCallable()`
to really evaluate the current port state on every repeat
* Add finite `100` repeat number to avoid infinite blocking when proxy is not started at all
* Add `doOnError()` for proxy `Mono` to log `ERROR` when repeat is exhausted
* * Fix Checkstyle violation
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Oliver <oli-ver@users.noreply.github.com>
