Add a reply timeout hook to TCP inbound gateway #3107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Closed

garyrussell opened this issue Nov 11, 2019 · 2 comments

Assignees

Labels

in: TCP/UDP type: enhancement

Milestone

Contributor

garyrussell commented Nov 11, 2019 •

edited by artembilan

Loading

Allow the user to determine the action after a reply times out.

https://stackoverflow.com/questions/58796500/tcpinboundgateway-server-configuration-on-timeout-a-custom-message-to-be-sent

garyrussell added type: enhancement in: TCP/UDP labels

garyrussell added this to the Backlog milestone

Member

artembilan commented Nov 12, 2019

I wonder if this feature could help us here:

/**
 * If errorOnTimeout is true, construct an instance that will send an
 * {@link ErrorMessage} with a {@link MessageTimeoutException} payload to the error
 * channel if a reply is expected but none is received. If no error channel is
 * configured, the {@link MessageTimeoutException} will be thrown.
 * @param errorOnTimeout true to create the error message.
 * @since 4.2
 */
public MessagingGatewaySupport(boolean errorOnTimeout) {

Looks like only HTTP one implement this logic when it is in a gateway mode:

public BaseHttpInboundEndpoint(boolean expectReply) {
	super(expectReply);
	this.expectReply = expectReply;
}

So, it looks like with that MessageTimeoutException and an errorChannel we have a hook to handle missed replies during timeouts.

Contributor Author

garyrussell commented Nov 12, 2019

Yes, it looks like that would work - the "hook" to return some other response can be done on the error flow.

artembilan self-assigned this

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

7fc04ed

Fixes spring-projects#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

artembilan mentioned this issue

GH-3107: Add errorOnTimeout for Inbound Gateways #3108

Closed

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

Fixes spring-projects#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

garyrussell closed this as completed in

72f7c72

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

fa6119d

Fixes spring-projects#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

artembilan added a commit to artembilan/spring-integration that referenced this issue


          Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

13bd6a0

…way"

This reverts commit fa6119d.

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

cfac0b4

Fixes spring-projects#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

artembilan added a commit to artembilan/spring-integration that referenced this issue


          Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

15d2628

…way"

This reverts commit fa6119d.

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

d77f06e

Fixes spring-projects#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

artembilan added a commit to artembilan/spring-integration that referenced this issue


          Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

…way"

This reverts commit fa6119d.

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

c2779df

Fixes spring-projects#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

artembilan added a commit to artembilan/spring-integration that referenced this issue


          Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

a5ee095

…way"

This reverts commit fa6119d.

artembilan added a commit to artembilan/spring-integration that referenced this issue


          spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGateway

366b9bf

Fixes spring-projects#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

artembilan added a commit to artembilan/spring-integration that referenced this issue


          Revert "spring-projectsGH-3107: Add errorOnTimeout for TcpInboundGate…

66480b6

…way"

This reverts commit fa6119d.

garyrussell pushed a commit that referenced this issue


          Use EmitterProcessor in the FluxMessageChannel (#3104)

94e0816

* 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 #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 fa6119d.

* * 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

artembilan modified the milestones: Backlog, 5.2.2

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment