* Use EmitterProcessor in the FluxMessageChannel
The `EmitterProcessor` has a good logic to block upstream producer
when its downstream subscriber cannot keep up with overproducing.
* Rework `FluxMessageChannel` logic to rely on the `EmitterProcessor`
instead of `Flux.create()`
* Cancel `FluxMessageChannel` internal subscriptions in the `destroy()`
* Fix `ReactiveStreamsTests.testFluxTransform()` for the splitter's
delimiter
* Ensure in the `FluxMessageChannelTests.testFluxMessageChannel`
that we can have several concurrent subscribers to the
`FluxMessageChannel`
* * Use `flux.onComplete()` instead of iteration over subscribers
* Change `subscribers` list into just `AtomicInteger` count marker
* fix `DefaultSplitterTests` according a new logic in the `FluxMessageChannel`
* GH-3107: Add errorOnTimeout for TcpInboundGateway
Fixes https://github.com/spring-projects/spring-integration/issues/3107
The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout
* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option
* * Use `delaySubscription()` for subscribing publishers in the `FluxMessageChannel`
to wait until this one subscribed.
* Use an `EmitterProcessor` to catch subscriptions and pass them as a
signal to delayed upstream publishers
* Fix `FluxMessageChannelTests.testFluxMessageChannelCleanUp` to
verify an actual property instead of removed.
* Fix `RSocketOutboundGatewayIntegrationTests` for the proper subscription
into a `FluxMessageChannel` before actual interaction with an RSocket
gateway.
This should help us also to avoid some race conditions in the future
* Revert "GH-3107: Add errorOnTimeout for TcpInboundGateway"
This reverts commit fa6119ddc4e423e8b15ca3fdc9fc4c79ea9d56af.
* * Refactor `FluxMessageChannel` to use `ReplayProcessor` for `subscribedSignal`.
This one is used `delaySubscription` for the upstream publishers
* Use a `AtomicBoolean` for subscription state since `doOnSubscribe()`
is called before `EmitterProcessor` adds subscribers for its `downstreams`
* Use `publishOn(Schedulers.boundedElastic())` for upstream publishers
to avoid blocking over there when our `EmitterProcessor` doesn't have
enough demand
* Refactor reactive tests to have a subscription into the `FluxMessageChannel`
earlier than emission happens for it
* * Use `Flux.subscribe(Consumer)` instead of `doOnNext(Consumer).subscribe()`
* * Emit `subscribedSignal` value after `.subscribe(subscriber)`
instead of `doOnSubscribe`
* Check for `this.processor.hasDownstreams()` before emitting such an event
* * Use `this.processor.hasDownstreams()` as a value to emit for `subscribedSignal`.
This way we are less vulnerable race conditions when subscribers are changed
actively
* Introduce RSocketInteractionModel
* Deprecate `RSocketOutboundGateway.Command` in favor of newly introduced
top level `RSocketInteractionModel`
* Deprecate setter, DSL and XML configurations for the deprecated
`RSocketOutboundGateway.Command` in favor of newly introduced configurators
for the mentioned `RSocketInteractionModel`
* Add `IntegrationRSocketEndpoint.getInteractionModels()` contract for
inbound endpoints.
This way we can restrict mapped endpoints to the particular interaction
model(s)
* Add DSL and XML configuration for inbound interaction model option
* Document changes
* * Rebased into master; fix conflicts
* Fix consistency for `RetrieveSpec` in the `RSocketOutboundGateway`
and `rsocket.adoc`
* * Fix language in JavaDos and XSD attributes
Fixes https://github.com/spring-projects/spring-integration/issues/3114
The contract of SpEL with its
`getValue(EvaluationContext context, @Nullable Object rootObject)` is
that we need to deal with provided `rootObject` even if it is `null`
and don't consult with `context.getRootObject()`
* Fix `ExpressionEvalMap` to have an internal `rootExplicitlySet`
to indicate that `root` explicitly provided by consumer, even if it is null.
According this flag call respective `Expression.getValue()`
* Add `@Nullable` to methods and their arguments into `ExpressionEvalMap`
& `FunctionExpression` to honor `Expression` contracts
* Populate an `HttpEntity` explicitly into `ExpressionEvalMap` from the
`HttpRequestHandlingEndpointSupport` and `WebFluxInboundEndpoint` for
full picture
* Fix JavaDocs indentations in the `ExpressionEvalMap`
Related to https://github.com/spring-projects/spring-framework/issues/23999
* Since Spring Integration inbound endpoints are generic in their method
signature we can't rely on a new `EMPTY_CONDITION` because it turns on
configuration merge into just `FrameType.REQUEST_FNF` &
`FrameType.REQUEST_RESPONSE`.
So, use `FrameType.REQUEST_FNF`, `FrameType.REQUEST_RESPONSE`,
`FrameType.REQUEST_STREAM` & `FrameType.REQUEST_CHANNEL` explicitly to
cover all the valid request-response models for SI endpoints
* Rework `RSocketOutboundGatewayIntegrationTests` according new logic.
Plus refactor to earlier subscription into `FluxMessageChannel` to
avoid potential race conditions
* INT-3494: Resolve dir for writing as a Resource
JIRA: https://jira.spring.io/browse/INT-3494
The expression for local directory can be resolved into a `Resource`
or resource location.
* Fix `ExpressionUtils.expressionToFile()` to support `Resource` and
also use `ResourceUtils.getFile(path)` when expression result is a string
* Modify tests to ensure that resource is resolved properly
* Upgrade affected tests to JUnit 5
* Mention an new functionality in docs
* * Improve Java doc for `ExpressionUtils.expressionToFile()`
* Finish the sentence in the `file.adoc`
Fixes https://github.com/spring-projects/spring-integration/issues/3107
The `MessagingGatewaySupport` has an `errorOnTimeout` option to throw
a `MessageTimeoutException` when downstream reply doesn't come back in
time for configured reply timeout
* Expose an `errorOnTimeout` option as a `TcpInboundGateway` ctor
property
* Add new factory methods into a `Tcp` factory for Java DSL
* Ensure a property works as expected in the `IpIntegrationTests`
* Document a new option
* Add a setter for MessagingGatewaySupport.errorOnTimeout option
* Expose an `errorOnTimeout` option on the DSL's `MessagingGatewaySpec`
making all the out-of-the-box inbound gateways possible to react to the
`MessageTimeoutException` when no reply during reply timeout
* Propagate properly `errorOnTimeout` in the `JmsInboundGateway`
* Modify docs respectively
* Improve docs about `errorOnTimeout`
Related to https://github.com/spring-projects/spring-framework/issues/23812
* Modify `FunctionsTests.kt` test to be sure that fall back to SpEL
invocation in case of Kotlin lambda works as expected.
This is like an additional test to the fix in Spring Framework
* Currently `MessagingMethodInvokerHelper.getTargetClass()` logs a
debug message when it can't cast a `targetObject` into an `Advised`.
Such a message is misleading and really unnecessary noise in the target
applications
* Fix the logic of the method to check `targetObject instanceof Advised`
before trying to extract interfaces from proxy.
This way we don't need to catch any exceptions at all.
If something will fail, it should be bubbled upstream for possible fixes
* Convert `MethodInvokingMessageProcessorTests` to JUnit 5 and restore
a couple tests
Fixes https://github.com/spring-projects/spring-integration/issues/3105
The loop in the `AbstractRemoteFileStreamingMessageSource` doesn't check
the next polled file for filtering.
* Add `continue;` when we filter the current file and poll the next one.
So, we go over into the `while()` beginning
* Clean up the code style and cover more code path in test
Fix a logic in the `JsonToObjectTransformer.obtainResolvableTypeFromHeadersIfAny`
when we have both `JsonHeaders.RESOLVABLE_TYPE` and `JsonHeaders.TYPE_ID`
and `JsonHeaders.RESOLVABLE_TYPE` header is valid `ResolvableType`, so
`JsonHeaders.TYPE_ID` doesn't override its value to return
* GH-3096: Skip RESOLVABLE_TYPE header in mapping
Fixes https://github.com/spring-projects/spring-integration/issues/3096
When we sent an AMQP message we should not map a
`JsonHeaders.RESOLVABLE_TYPE` header which is a `ResolvableType` and
isn not compatible after converting to string
Also improve `JsonToObjectTransformer` to ignore a
`JsonHeaders.RESOLVABLE_TYPE` when it is type of String
* * Fix `obtainResolvableTypeFromHeadersIfAny()` logic
* Use `EmitterProcessor` for Channels adaptation
Related https://github.com/spring-cloud/spring-cloud-stream/issues/1835
To honor a back-pressure after `MessageChannel` adaptation it is better
to use an `EmitterProcessor.create(1)` instead of `Flux.create()`.
This way whenever an emitter buffer is full, we block upstream producer
and don't allow it to produce more messages
**Cherry-pick to 5.1.x**
* * Wrap every new subscription into a `Flux.defer()`
* Fix `ReactiveStreamsConsumerTests` to use a new `Subscription` after
each `stop()/start()` on the `ReactiveStreamsConsumer`
* * Remove unused imports
Related to: https://github.com/spring-projects/spring-boot/issues/18812
* Extract `ServerRSocketMessageHandler` into a `public` class to allow
to configure it as top-level bean and bind it into an existing server
* Change `ServerRSocketConnector` to accept the mentioned external bean
and don't create an internal RSocket server with an assumption that it
is create externally
* Add `IntegrationRSocketMessageHandler.requestMappingCompatible`
option to allow to configure `ServerRSocketMessageHandler` for both
Spring Integration RSocket channel adapters and regular `@MessageMapping`.
This is useful for Spring Boot auto-configuration
These changes give a hook to auto-configure Spring Integration RSocket
channel adapters in Spring Boot
* GH-3089: Add AmqpInGateway.replyHeadersMappedLast
Fixes https://github.com/spring-projects/spring-integration/issues/3089
In some use-case we would like to control when headers from SI message
should be populated into an AMQP message.
One of the use-case is like a `SimpleMessageConverter` and its `plain/text`
for the String reply, meanwhile we know that this content is an
`application/json`.
So, with a new `replyHeadersMappedLast` we can override the mentioned
`content-type` header, populated by the `MessageConverter` with an
actual value from the message headers populated in the flow upstream
* Introduce an `AmqpInboundGateway.replyHeadersMappedLast`; expose it
on the DSL and XML level
* Use newly introduced `MappingUtils.mapReplyMessage()`
* Optimize `DefaultAmqpHeaderMapper` to not parse JSON headers at all
when `JsonHeaders.TYPE_ID` is already present (e.g. `MessageConverter`
result)
* Also skip `JsonHeaders` when we `populateUserDefinedHeader()`
**Cherry-pick to 5.1.x**
* * Fix language and package typos
* Add missed `@param` in JavaDoc of the `AmqpBaseInboundGatewaySpec.batchingStrategy()`
* Extract a `RabbitTemplate` `MessageConverter` to use for reply messages
conversion - pursue a backward compatibility
* GH-3090: Add `logout() to `FtpSession.close()`
Fixes https://github.com/spring-projects/spring-integration/issues/3090
Without `logout()` the FTP session is not closed at all,
but just the connection is closed.
Some FTP servers close those sessions eventually anyway, but some just
leak with resources.
**Cherry-pick to 5.1.x & 4.3.x**
* * Migrate `SessionFactoryTests` to JUnit 5
Fixes https://github.com/spring-projects/spring-integration/issues/3026
**Cherry-pick to `5.1.x`**
* Populate proper `FileTransferringMessageHandler` impl from DSL spec
implementations.
This way we are able to use a provided `chmod` from Java DSL
* Added `FileTransferringMessageHandlerSpec` ctor TODO
* Update SftpTests
* Code cleanup; `@Ignore` `SftpTests.testSftpOutboundFlowWithChmod()`
since it doesn't work properly on Windows
* Extract `BaseIntegrationFlowDefinition`
Kotlin inline reified functions cannot override as syntax sugar generic
Java methods.
* Extract a `BaseIntegrationFlowDefinition` with strong typed methods,
plus those which doesn't have a syntax sugar variants
* Leave in the `IntegrationFlowDefinition` only those generic methods
which are just a syntax sugar without any types requirements
This way we can write Kotlin inline reified functions as an extension
with the same names to make a Kotlin DSL much cleaner
* * Override methods in the IntegrationFlowDefinition for bytecode compatibility
* Extract `protected` getters and setters for `BaseIntegrationFlowDefinition`
properties to allow a proper extension
* Use those getters and setters in the code
* * Add // NOSONAR for overridden methods
