Always assumed import FlowGraph.Implicits._ and all other necessary imports:
I can pack all my logic into a function:
val f: FlowGraph.Builder => Outlet[T] => Outlet[S]
for some custom type T and type S. Now I can wrap/use this function as a
Flow or PartialFlowGraph:
val flow: Flow[Input, Trigger.type, Unit] = Flow() { implicit b =>
val broadcast = b.add(Broadcast[Input](1))
(broadcast.in, f(b)(broadcast.out(0)))
}
or
val partialFlowGraph = FlowGraph.partial() { implicit b =>
val broadcast = b.add(Broadcast[Input](1))
UniformFanInShape(f(b)(broadcast.out(0)), broadcast.in)
}
Now in my main flowgraph:
FlowGraph.closed() { implicit b =>
val in = Source.empty[T]
val out = Sink.ignore
in ~> flow ~> out
// or :
val fg = b.add(partialFlowGraph)
in ~> fg.in(0)
fg.out ~> out
}
In reality the above flowgraph is more complex but for this showcase a
dummy in and out is used, e.g. actually "in" is a PropsSource[T] externally
feeded by an actor.
I want flow or partialFlowGraph to have zero buffers internally but at the
outside as usual and thus behave as a single Flow[T].map(f).
So to answer your first question: "should act like a single stream stage"
from the perspective of the main flowgraph.
And my problem (second question) is that I have a PropsSource[T] feeding my
main flowgraph which buffers external incomming events with a custom
bufferlogic.
A example will make that clear:
Lets assume for the moment this events are timestamps every 30ms and the
processing of these timestamps by my function f respectively
flow/partialFlowGraph takes more than 30ms (time for all internal stages
together for one element)
Then the buffers of all involved stages in main and especially in my
partialFlowGraph get filled up over time.
I like to discard old timestamps in my PropsSource-Bufferlogic in that
case. But the old timestamps are stucked in the processing chain of my
partial flowgraph.
So in.buffer(1, OverflowStrategy.dropHead) won't help here.
Hope this simplified example shows my need for something like FuseStage.
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
