Hi Vladimir,
you have hit on the interesting issue. Your understanding is correct but
the output of your example is not what you expect, because every stage in
the stream works in batches. The size of the batch is the size of the input
buffer which can be set in *akka.stream.materializer.{initial,
max}-input-buffer-size.*
So when zip asks upstream for more elements it asks for more than one
element. That request can be fulfilled by the faster source with an element
every 100 milliseconds which it does and conflate does not have time to
kick in. That is why you see lots of 1s printed out.
>From time to time buffers align and zip's request for more elements is
fulfilled first by accumulated value by the conflate and then again by 1s
every 100 milliseconds.
A quick fix is setting input buffer size to 1 for a whole stream. However
that makes the whole stream slower.
val set = MaterializerSettings(system).withInputBuffer(1, 1)
implicit val fm = FlowMaterializer(set)
A more proper fix would be to provide "synchronize" element which would be
the same zip under the hood but with the buffer size of one. Could you
create an issue in Akka issue tracker for that?
On Tue, Oct 14, 2014 at 11:07 PM, Vladimir Koshelev <
[email protected]> wrote:
> 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.
>
--
Martynas Mickevičius
Typesafe <http://typesafe.com/> – Reactive
<http://www.reactivemanifesto.org/> Apps on the JVM
--
>>>>>>>>>> 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.
