Add Reactive mode for AbstractPollingEndpoint #2429
Conversation
|
There is still need to implement such a mode for the |
|
Would be great to see your opinion @smaldini ! 😄 |
|
any chances that this can make it into our current I mean that I had recently a question about Reactive JMS on Gitter: https://gitter.im/spring-projects/spring-integration?at=5b9644fef3c26b08f65608d4 This Reactive polling adaption will do for us a trick when we won't ask broker for more messages until downstream demand. I believe we can do something similar with the event-driven channel adapters when we will block consuming thread until demand from the downstream. Different story though... Thanks |
|
Sounds like a plan if it can make the RC1 on 9/20. |
* When `SourcePollingChannelAdapter.outputChannel` is a `ReactiveStreamsSubscribableChannel`, use `Flux.generate()` for polling * Refactor `AbstractPollingEndpoint` to remove redundant `Poller` class in favor of lambda * Extract `pollForMessage()` method to handle TX states instead of `Poller` class previously
artembilan
force-pushed
the
ReactivePoller
branch
from
119975c to
6827476
Compare
|
So, Gary, can we merge it now or you would like to see anything else here? I would address Docs in the separate PR somewhere in between RC and GA. Thanks |
| @@ -66,23 +74,27 @@ | |||
|
|
|||
| private boolean syncExecutor = true; | |||
|
|
|||
| private ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); | |||
|
|
|||
| private Trigger trigger = new PeriodicTrigger(10); | |||
There was a problem hiding this comment.
Default 10ms? Really?
If we must have a default, it should be much larger than this.
Why wouldn't we just say it's invalid to configure a reactive poller without a trigger?
There was a problem hiding this comment.
Huh? This is an existing code. I just reformatted it a bit to remove some volatile.
We can reconsider the default in other issue, although I think Mark did this way to make as close to the event-driven behavior as possible
SourcePollingChannelAdapter.outputChannelis aReactiveStreamsSubscribableChannel, useFlux.generate()for pollingAbstractPollingEndpointto remove redundantPollerclassin favor of lambda
pollForMessage()method to handle TX states instead ofPollerclass previously