It can be convenient to attach metadata to each element in the stream.
FlowWithContext is just a
Flow that contains tuplespairs of element and context, but the advantage is in the operators: most operators on
FlowWithContext will work on the element rather than on the tuplepair, allowing you to focus on your application logic rather without worrying about the context.
Not all operations that are available on
Flow are also available on
FlowWithContext. This is intentional: in the use case of keeping track of a read offset, if the
FlowWithContext was allowed to arbitrarily filter and reorder the stream, the
Sink would have no way to determine whether an element was skipped or merely reordered and still in flight.
For this reason,
FlowWithContext allows filtering operations (such as
collect, etc.) and grouping operations (such as
sliding, etc.) but not reordering operations (such as
statefulMapConcat). Finally, also ‘one-to-n’ operations such as
mapConcat are allowed.
Filtering operations will drop the context along with dropped elements, while grouping operations will keep all contexts from the elements in the group. Streaming one-to-many operations such as
mapConcat associate the original context with each of the produced elements.
As an escape hatch, there is a
via operator that allows you to insert an arbitrary
Flow that can process the tuplespairs of elements and context in any way desired. When using this operator, it is the responsibility of the implementor to make sure this
Flow does not perform any operations (such as reordering) that might break assumptions made by the
Sink consuming the context elements.
The simplest way to create a
SourceWithContext is to first create a regular
Source with elements from which the context can be extracted, and then use Source.asSourceWithContext.
When you have a
source that produces elements of type
Foo with a context of type
Ctx, and a
Bar, you cannot simply
source.via(flow) to arrive at a
SourceWithContext that produces elements of type
Bar with contexts of type
Ctx. The reason for this is that
flow might reorder the elements flowing through it, making
via challenging to implement.
Due to this there is a
unsafeDataVia that can be used instead however no protection is offered to prevent reordering or dropping/duplicating elements from stream so use this operation with great care.
There is also a Flow.asFlowWithContext which can be used when the types used in the inner
Flow have room to hold the context. If this is not the case, a better solution is usually to build the flow from the ground up as a
FlowWithContext, instead of first building a
Flow and trying to convert it to