36 fix streams drain control #37
Conversation
core/src/test/scala/com/softwaremill/kmq/redelivery/IntegrationTest.scala
Outdated
Show resolved
Hide resolved
core/src/test/scala/com/softwaremill/kmq/redelivery/IntegrationTest.scala
Outdated
Show resolved
Hide resolved
core/src/main/scala/com/softwaremill/kmq/redelivery/streams/RedeliveryTrackerStream.scala
Outdated
Show resolved
Hide resolved
mkrzemien
force-pushed
the
36-fix-streams-drain-control
branch
3 times, most recently
from
6d766fd to
7a0c34d
Compare
mkrzemien
force-pushed
the
36-fix-streams-drain-control
branch
from
7a0c34d to
e4ffc31
Compare
| RunnableGraph | ||
| .fromGraph(GraphDSL.createGraph(redeliverySink, commitMarkerSink)(combineFutures) { | ||
| val cancellable = RunnableGraph | ||
| .fromGraph(GraphDSL.createGraph(redeliverySink, commitMarkerSink)(Keep.left) { |
There was a problem hiding this comment.
don't we need to cancel both sinks?
There was a problem hiding this comment.
What's special in the RedeliverySink is an additional TickSource that is instantiated there. Without the TickSource both sinks appear to be closed correctly and the DrainingControl is able to shut down the stream.
Here I'm exposing a hook to the TickSource as Cancellable materialized view from the whole RedeliverySink.
| .toMat(Sink.ignore)(DrainingControl.apply) | ||
| .run() | ||
| } | ||
| } | ||
|
|
||
| class AggregatingToMatFlow[T] extends GraphStageWithMaterializedValue[FlowShape[T, T], Iterable[T]] { |
There was a problem hiding this comment.
I hope it's interesting in a positive way :)
It's certainly not the simplest in general. But here we have additional TickSource merged inside one of the nested sinks which doesn't stop when it's source stops.
There was a problem hiding this comment.
Yes, in a positive way :)
I think the problem of stopping sub-streams is not that uncommon, as it emerges also when using the .groupBy which I mentioned. So either it's already solved in some way, or we have a solution which would be a great material for a short blog post :)
Closes #36