Fixes https://github.com/spring-projects/spring-integration/issues/3497
The `BaseIntegrationFlowDefinition.REFERENCED_REPLY_PRODUCERS` is filled up
only with the `MessageProducer` instances.
Therefore no reason to calculate a hash code for every single bean passed
to the `ReplyProducerCleaner.requiresDestruction()`
* Check for the `MessageProducer` instance before passing the bean to the
`BaseIntegrationFlowDefinition.REFERENCED_REPLY_PRODUCERS.contains()`
* Check for the `MessageProducer` in the `ReplyProducerCleaner.postProcessBeforeDestruction()`,
too.
It can be called independently of the `requiresDestruction()` and will calculate a hash code
from the bean again for nothing
**Cherry-pick to 5.4.x & 5.3.x**
* Clean up code style in Gradle configs
* Fix JavaDoc errors
* Make JUnit 4 as `optional` dependency for `spring-integration-test-support`
* Remove redundant `javax.activation-api` from some modules
Having a bean as a `java.util.Properties` is not confusing
and may lead to some conflicts in the real application.
Plus it is not so easy to configure: need to know all
the possible integration properties - bad end-user experience
* Make an `IntegrationProperties` as a public POJO
for easy configuration of global properties
* Deprecate a presence of the `java.util.Properties`
* Leave framework-created `integrationGlobalProperties` as
a `Properties` instance for backward compatibility
* Fix tests to expose an `IntegrationProperties` bean instead
of deprecated `Properties`
* Fix docs according the change and current recommendations
The logic in the `FluxMessageChannelTests.testFluxMessageChannelCleanUp()`
relies on the `finishLatch.countDown()` which happens before we
release the lock for a group in the aggregator.
So, when we call a `destroy()` as the next statement, we interrupt
the lock instead of unlocking.
This cause a race condition in the Reactor's `Sink` to terminate properly
* Expose `CorrelationHandlerSpec.releaseLockBeforeSend()` and set it to `true`
in the test to properly unlock before we call `finishLatch.countDown()`
* Deprecate a couple introduced before options in favor of their variants
with `set` prefix
* Fix new Sonar smells
* GH-3464: Treat 0 as special for polling endpoint
Fixes https://github.com/spring-projects/spring-integration/issues/3464
The `maxMessagePerPoll <= 0` is considered as an unbound `receive()` call.
End-users would like to have a special treatment for `0` value -
skip the `receive()` call altogether for the current polling cycle.
* Change the logic for a scheduled poller to not call `pollForMessage()`
and just log an INFO when `maxMessagePerPoll == 0`
* Fix reactive poller to deal with `maxMessagePerPoll == 0` properly
* Expose `maxMessagesPerPoll` as a `@ManagedAttribute` to let it to be
modified via Control Bus and JMX
* Test and document a new behavior
* * Fix unused imports in the test class
* Fix language in docs accoridng PR review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
The current `Flux.take()` doesn't allow an arg `< 0` treating
it as an unbound request.
* Change `take()` to `limitRequest()` according strict `MessageSource.receive()`
producing expectations
* Treat `maxMessagesPerPoll < 0` as a `Long.MAX_VALUE` for unbound requests;
`0` is treated in the `limitRequest()` as "no more requests - cancel"
* Revise `AbstractPollingEndpoint` for `LogAccessor` usage
* Add `AbstractPollingEndpoint` class JavaDocs
* Fix tests according `AbstractPollingEndpoint` changes
**Cherry-pick to `5.4.x`**
* Expose `error.ignoreFailures` global property
The global default `errorChannel` can have more then only one logging subscriber.
There is no guarantee in which order subscribers are called and one of them fails
the rest are not called.
* Expose `spring.integration.channels.error.ignoreFailures` global property for
default `errorChannel` and make it `true` by default for better end-user experience
when it is expected to have error being handled by all the subscribers
* * Fix language in the `whats-new.adoc`
* * Fix `JmsTests` for error channel to propagate an exception
* More language fixes for whats-new.adoc
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
Fixes https://github.com/spring-projects/spring-integration/issues/3473
When `AbstractEndpoint.start()` and `AbstractEndpoint.isRunning()`
are called from different thread on `synchronized` methods,
we may end up with a dead lock: one thread waits for monitor on
`synchronized` and another waits for the `lifecycleLock`
* Change `AbstractEndpoint.isRunning()` to a plain `return this.running;`
- there is no reason in a lock around returning this `volatile` property state
**Cherry-pick to 5.4.x & 5.3.x**
Fixes https://github.com/spring-projects/spring-integration/issues/3474
The `AbstractTransformer` implementation just wraps a thrown exception into
a `MessageTransformationException`.
It is expected with any method transformer impl (POJO, Lambda direct etc.)
the exception is thrown as is without extra wrapping into an `IllegalStateException`
for consistency
* Fix `LambdaMessageProcessor` to rethrow `RuntimeException` as is
and only wrap into an `IllegalStateException` for all other exceptions
* Fix tests for missed extra stack trace
* Copy request headers when reply is not a Message
Related to https://stackoverflow.com/questions/65887787/spring-integration-unexpected-behavior-on-error-handling-in-splitter-aggregator
Typically `AbstractMessageProducingHandler` implementation with the `shouldCopyRequestHeaders = false`
returns a `Message`(or `MessageBuilder`) as a reply, but when some request handler `Advice` is involved,
we may produce a plain payload and the reply message does not have request headers
* Improve `AbstractMessageProducingHandler` logic to copy request headers into a reply message builder,
when the original reply is not a `Message` or `MessageBuilder`.
This way we improve end-user experience and set them free from copying those headers manually
* Justify a new behavior with the `TransformerTests.testFailedTransformWithRequestHeadersCopy()`
* * Fix Checkstyle violations
Fixes https://github.com/spring-projects/spring-integration/issues/3446
* For better resources utilization provide a `Stream<Message<?>>` API
on the `MessageGroupStore`, `MessageGroup` and `MessageGroupQueue`
* Use this API in the `DelayHandler` when it reschedules persisted messages
* add `Comparable` contract to the initial wrapper so that expressions
containing selection/projection with JsonNode filtering are now possible
* render directly the value of a ValueNode. SpEL parser can now evaluate
correctly the object value when the expression deals with filtering on
`JsonNode` values (<, >, ==...)
* the 2 above changes make this possible:
`property.^[name.getTarget().asText() == 'value1'].name` can now be
written as simply as `property.^[name == 'value1'].name`. The expression
is now fully compatible with other PropertyAccessor types
* rename `ToStringFriendlyJsonNode` to `JsonNodeWrapper` as it reflects
more the capability of the new wrapper
* add a GenericConverter to be able to convert the `JsonNodeWrapper`
class into a `JsonNode` (or its derivative classes) directly while
calling `expression.getValue(..., JsonNode.class)`
* add the possibility to access list items in a json-path like index
=> a negative number will begin from the end of the list <=
* return null value when `JsonNode` cannot handle the property name
(or index)
* Change Json Converter
* Clean up the code style
* Convert JsonPropertyAccessor tests to JUnit 5
* GH-3450: Expose PubSub requireSubscribers option
Fixes https://github.com/spring-projects/spring-integration/issues/3450
* Add `PublishSubscribeChannel` ctors with the `requireSubscribers` option
with a direct delegation to the same option of underlying `BroadcastingDispatcher`
* Add factory methods to the `Channels` & `MessageChannels` to expose
this new `requireSubscribers` option
* Expose an XML `require-subscribers` attribute for the `<publish-subscribe-channel`
* Document this new `requireSubscribers` option
* Introduce a global `spring.integration.channels.error.requireSubscribers` property
for a default `errorChannel`
* Document this property and explain its default `true` in the `error-handling.adoc`
* Remove docs for `spring.integration.postProcessDynamicBeans` since it purpose was
removed since version `5.1` in favor of "always post-process behavior" as it was
always with all the `BeanPostProcessor`s
* * Fix language in docs according review
* * Add more docs about `requireSubscribers` and cross-links between chapters
* GH-3452: Warn Publisher.onError in NullChannel
Fixes https://github.com/spring-projects/spring-integration/issues/3452
The `NullChannel` can subscribe to the `Publisher` payload fully ignoring
the possible data according the `NullChannel` nature.
However an error thrown from the reactive stream processing is also ignored.
* Log WARN message from the `Subscriber.onError()` when `NullChannel`
subscribes to the `Publisher` produced to this channel.
* Mention the logic in the `NullChannel` docs; point to the
`ReactiveRequestHandlerAdvice` for further possible error handling in
the target application
**Cherry-pick to `5.4.x`**
* * Remove unused imports
* * Fix `PseudoTransactionalMessageSourceTests` for `LogAccessor`
from `NullChannel`
* Subscribe to Publisher in the NullChannel
If a payload of the message in the flow is a `org.reactivestreams.Publisher`,
it must be subscribed somewhere downstream to initiate reactive processing.
The `NullChannel` just ignores the message altogether and therefore `Publisher`
is lost
* Check the payload of the message in the `NullChannel` for `Publisher` type
and subscribe to it
* Verify in the test and mention this logic in the docs
**Cherry-pick to 5.4.x**
* Apply suggestions from code review
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix compatibility with upgraded dependencies
* Move `What's New` for `5.4` to `history`
* Add `XmlUnit` dependency since the latest AssertJ has deprecated its `isXmlEqualTo()`
* Remove `javax.annotation-api` and `apiguardian-api` as redundant dependencies
Related to https://build.spring.io/browse/INT-MASTER-2240
Turns out the `doOnRequest()` is still not enough to be sure
that subscriber is accepted into the `Publisher`.
Probably because `doOnRequest()` maybe really asked for the
subscription by itself for some logic
* Use `Mono` with `repeatWhenEmpty()` until
`this.sink.currentSubscriberCount() > 0`
Related to https://build.spring.io/browse/INT-MASTERSPRING40-JOB1-1271
* Fix race condition with subscription in the `FluxMessageChannel`:
rely on the `doOnRequest()` to start producing from upstream publishers
* Remove `SourcePollingChannelAdapterTests` since
`Class.getDeclaredConstructor().newInstance()` doesn't rethrow an exception as is
but wrap it into the `InvocationTargetException`.
This is probably the main reason to deprecate a regular `Class.newInstance()`
* Use `LettuceConnectionFactory.setEagerInitialization(true)` in the `RedisAvailableRule`
to avoid possible dead lock with lazy init
* Adjust Redis tests for new `RedisAvailableRule` behavior
The `IntegrationManagementConfigurer` is a `BeanPostProcessor`
so it must not have direct dependency injection for other beans.
In our case it is a `MetricsCaptor` injected from the
`IntegrationManagementConfiguration`
* Fix `IntegrationManagementConfiguration` and `IntegrationManagementConfigurer`
to rely on the `ObjectProvider<MetricsCaptor>` instead
Tested against latest Spring Boot
**Cherry-pick to `5.3.x` & `5.2.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3425
It turns out that `IntegrationManagementConfigurer.destroy()` is still too
late to remove `gauges` and Micrometer tries to gather them on application
context close, when many beans are already destroyed
* Catch `ContextClosedEvent` in the `IntegrationManagementConfigurer`
and removed `gauges` from there
* Remove `destroy()` impl since it is out of use already:
the `IntegrationManagementConfigurer` is not supposed to be in the
target application directly, so it looks safe to remove the `DisposableBean`
altogether
**Cherry-pick to `5.3.x` & `5.2.x`**
* Rely on `MProducerSupport.active` for `Flux`
* Fix `MessageProducerSupport` to extract an `active` flag and set it before
`isRunning` - the `Flux` subscription relies on the `takeWhile()`
where in case of `autoStartup = false` we will never start consume because
it is set to `true` already after `doStart()`
* Refactor all the `MessageProducerSupport` implementation with similar
`active` state to use already one from the super class
**Cherry-pick to 5.3.x**
* * Remove `MessageProducerSupport.setActive()`
to not let to mutate it from the implementations
* Set `active` to `false` in the `destroy()`
* Clean up and fix typos in the affected `JmsMessageDrivenEndpoint`
* * Pull `active` flag down to the `AbstractEndpoint`
* Set `active = true` in the `start()` before calling `doStart()`
* Do same for `active = false` in the `stop()`
* Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing
* Refactor endpoints to rely on the `active` state from the `AbstractEndpoint`
not their own
See SO for more info:
https://stackoverflow.com/questions/64456946/handle-exceptions-errors-other-than-messagingexception-ie-other-error-excepti
In the versions before `5.2.x` the `Error` was wrapped to the `MessagingException`
in the `AbstractRequestHandlerAdvice` and this one was handled properly
on the gateway level with its `errorChannel` configured
* Fix `MessagingGatewaySupport` to catch all the `Throwable` and try to send them as `ErrorMessage`
to the `errorChannel`.
Otherwise unwrap returned `MessagingException` and re-throw an `Error` as is to keep
the current behavior for non-handled exceptions
* Do not wrap `Error` into a `MessagingException` and re-throw as is if there is no `errorChannel`
**Cherry-pick to 5.3.x & 5.2.x**
With the `time-source` in Spring Cloud Stream we noticed
that first two items have the same (or close) value
when the `time-supplier` is used with the reactive output channel.
Turns out the `contactMap()` with `prefetch == 1` asks two
upstream elements immediately not waiting for a completion
for the fist one.
This way we were not able to update `triggerContext` for the proper
`lastCompletionTime`
* Change `AbstractPollingEndpoint` to `prefetch = 0` for the
"fair" upstream request
* Verify the behavior with the new
`ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()`
* Mark `ReactiveInboundChannelAdapterTests` as `@LongRunningTest`
since it is now pretty long waiting for all the 3 dates to verify
* Fix `InboundChannelAdapterAnnotationPostProcessor` to properly
register `MethodInvokingMessageSource` when we have more than one
`Supplier` with the `@InboundChannelAdapter` in the same configuration
class
If there is a `MeterRegistry` bean (any) in the application context,
we add a `MicrometerMetricsCaptor` bean which populates meters further
from the components.
In some case we may load a `MicrometerMetricsCaptor` bean too early
so, not all the stuff around `MeterRegistry` maybe ready.
See Spring Boot and its `MetricsAutoConfiguration`
* Fix the `MicrometerMetricsCaptorRegistrar` the way to rely on the
`ObjectProvider<MeterRegistry>` instead
* Add package protected ctor to the `MicrometerMetricsCaptor` to
provide a target `MeterRegistry` on demand
All of that will ensure that we use an already post-processed `MeterRegistry`
including Spring Boot auto-configuration
**Cherry-pick to 5.3.x & 5.2.x**
* Use 'onErrorContinue' to continue receiving messages from Stream
* Send an `ErrorMessage` to the provided `errorChannel` (if any)
* And `@Nullable` to some `MessageProducerSupport` API which definitely
may accept `null`
* Extract common `buildMessageFromRecord()` method in the `ReactiveRedisStreamMessageProducer`,
so all the headers from the stream `Record` are carried to the message independently of the
record state - normal send or error sending
Related to https://github.com/spring-projects/spring-framework/issues/25884
* Don't use `MonoProcessor` & `FluxProcessor` since they are deprecated now
* Fix RSocket components to use an `AtomicReference` header instead of deprecated
`MonoProcessor`
* Introduce an internal `IntegrationRSocketPayloadReturnValueHandler` to handle
properly a Spring Integration case for inbound requests with its
`IntegrationRSocketEndpoint` implementation
* Don't do response handling in the `RSocketInboundGateway` any more
since it is deferred now to the `IntegrationRSocketPayloadReturnValueHandler`
* Remove internal `ChannelSendOperator` since it is out of use from now on
Fixes https://github.com/spring-projects/spring-integration/issues/3306
To prevent API misuse (one can set ID after doing a registration which
will do nothing actually), it is safer to remove set* methods from
`StandardIntegrationFlowRegistration` object.
Providing some related refactoring like making fields 'final',
adjusting usages to keep correct behavior of the builder.
* 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>
The `Mono.doOnSuccess()` is always called for completed `MonoSink`
even if the value is `null`
* Fix `IntegrationReactiveUtils.messageSourceToFlux()` to check a
message for `null` before calling `AckUtils.autoAck()`
* Add an `logger.error()` message when `doOnError()` is invoked for that `Mono`
**Cherry-pick to 5.3.x**
* GH-3386: Add method signature for gateway proxy
Fixes https://github.com/spring-projects/spring-integration/issues/3386
All the gateway proxy method invokers are supplied with the same
bean name inherited from the proxy.
* Add method signature for proxy method bean to fine-grain the
management for those bean in the logs, message history and metrics
* * Use `Class.getSimpleName()` instead to make gateway names less verbose
* GH-3376: Remove gauges on application ctx close
Fixes https://github.com/spring-projects/spring-integration/issues/3376
The `MeterRegistry` may request meters on application shutdown.
The gauges for channels, handlers and message sources don't make sense
at the moment since all those beans are going to be destroyed.
* Remove gauges for channel, handler and message source numbers from the
`IntegrationManagementConfigurer.destroy()`
**Cherry-pick to 5.3.x & 5.2.x**
* * Add `MicrometerImportSelector` to conditionally load
a `MicrometerMetricsCaptorConfiguration` when `MeterRegistry`
is on class path.
* Make `MicrometerMetricsCaptorConfiguration.integrationMicrometerMetricsCaptor()`
bean dependant on the `ObjectProvider<MeterRegistry>`
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerImportSelector.class)`
* * Add `MicrometerMetricsCaptorRegistrar` to register a `MICROMETER_CAPTOR_NAME`
bean when `MeterRegistry` is on class path and no `MICROMETER_CAPTOR_NAME` bean yet.
* Make `IntegrationManagementConfiguration.managementConfigurer()`
dependant on the `ObjectProvider<MetricsCaptor>`.
This way the `IntegrationManagementConfigurer` is destroyed before
`MeterRegistry` when application context is closed
* Deprecate `MicrometerMetricsCaptor.loadCaptor()` in favor of
`@Import(MicrometerMetricsCaptorRegistrar.class)`
* Fix test to make a `MeterRegistry` bean as `static` since
`@EnableIntegrationManagement` depends on this bean definition now
* Fix unused import in the `IntegrationRequestMappingHandlerMapping`
* Fix deprecations from Reactor
* Fix race condition in the `AbstractCorrelatingMessageHandlerTests`:
the discard message is sent much earlier than group is removed from the store.
Iterate group count call until it pass or 10 seconds timeout
* Remove list size assert in the `FtpServerOutboundTests`: looks like
it is not updated properly even if we have an expected content in the collection
* Increase timeout to assert remote files removal in the `FtpRemoteFileTemplateTests`
* Use `tryEmitNext` on Reactor `Sink` since `emitNext` is deprecated
* Add `MessageDeliveryException` emission when `send()` returns `false`
in the `FluxMessageChannel` for `subscribeTo` provided `Publisher`
