Hi,
I'm trying to use akka-stream and struggling to understand how conflate
works. Here a snippet of code I'm using:
object ConflateDemo extends App {
import scala.concurrent.duration._
implicit val system = ActorSystem("demo")
implicit val fm = FlowMaterializer()
case object Tick
val fast = FlowFrom(Duration.Zero, 100.millis, () =>
Tick).conflate[Int](_ => 1, (acc,_) => acc + 1)
val slow = FlowFrom(Duration.Zero, 1.second, () => Tick)
import akka.stream.scaladsl2.FlowGraphImplicits._
val g = FlowGraph { implicit b =>
val zip = Zip[Int, Tick.type]
slow ~> zip.right
fast ~> zip.left
zip.out ~> FlowFrom[(Int, Tick.type)].map(_._1) ~> ForeachSink[Int]( v
=> print(" " + v))
}.run
}
}
My intention is to create two streams, first produces events 10 times
faster than the second one. Conflate on the fast stream should count
events. Then I zip those streams and print results of conflate out.
My understanding of how it should work:
- conflate consumes events produced by fast stream as fast as they can
be produced, so there is always demand and all ticks are emitted and
counted.
- As soon as an event from the slow flow arrives to the zip, zip demands
an element from conflate and sends them into sink.
So, I have expected output looking like: 10 10 10 10 10 ...
What I get: 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 55 1 1 1 1 1 1 1 73 1 1 1 1 1
1 1 73 1 1 1 1 1 1 1 73 1 1 1 1 1 1 1 73 1 1 1 1 1 1 1 72 1 1 1 1 1 1 1 74
1 1 1 1 1 1 1 72 1 1 ...
What am i missing? Is there some mistake in the code or do I misunderstand
how conflate and zip are working?
Vladimir
--
>>>>>>>>>> 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.
