val accumulator =
Flow[Int]
.statefulMapConcat { () ⇒
var total = 0
var curIdx = 0L
var startIdx = curIdx
i ⇒ {
val newTotal = total + i
if (newTotal >= 10) {
val result = List((newTotal, startIdx))
curIdx += 1
startIdx = curIdx
total = 0
result
} else {
val result = List.empty[(Int, Long)]
curIdx += 1
total = newTotal
result
}
}
}
On Wed, Nov 16, 2016 at 10:18 AM, Julian Howarth <[email protected]>
wrote:
> I have what seems a very simple problem but cannot come up with a solution
> for it.
> Firstly, I have a flow that does some accumulation, therefore only emits
> after it has received a number of elements. As a simple example:
>
> val accumulator =
> Flow[Int]
> .statefulMapConcat { () ⇒
> var total = 0
> i ⇒ {
> val newTotal = total + i
> if (newTotal >= 10) {
> total = 0
> List(newTotal)
> }
> else {
> total = newTotal
> List()
> }
> }
> }
>
>
> This works as expected:
>
> val result = Source(List(2, 5, 6, 3, 8, 4, 1, 8, 7, 4))
> .via(accumulator)
> .runWith(Sink.fold(Seq.empty[Int])(_ :+ _))
>
> result.futureValue shouldBe Seq(13, 11, 13, 11)
>
>
> Now, I want to record the index of the first element which started the
> accumulation. In my actual application this is a java.time.Instant, but
> Long suffices for this example. As a first naive attempt, using the Graph
> DSL:
>
> val flow =
> Flow[(Int, Long)]
> .via(Flow.fromGraph(GraphDSL.create() { implicit builder ⇒
> import GraphDSL.Implicits._
> val unzip = builder.add(Unzip[Int, Long]())
> val acc = builder.add(accumulator)
> val zip = builder.add(Zip[Int, Long]())
>
> unzip.out0 ~> acc ~> zip.in0
> unzip.out1 ~> zip.in1
>
> FlowShape[(Int, Long), (Int, Long)](unzip.in, zip.out)
> }))
>
> val result = Source(List(2, 5, 6, 3, 8, 4, 1, 8, 7, 4)).zipWithIndex
> .via(flow)
> .runWith(Sink.fold(Seq.empty[(Int, Long)])(_ :+ _))
>
> result.futureValue shouldBe Seq((13, 0), (11, 3), (13, 5), (11, 8))
>
>
> which times out as no elements flow through. So adding a buffer:
>
> val flow =
> Flow[(Int, Long)]
> .via(Flow.fromGraph(GraphDSL.create() { implicit builder ⇒
> import GraphDSL.Implicits._
> val unzip = builder.add(Unzip[Int, Long]())
> val acc = builder.add(accumulator)
> val zip = builder.add(Zip[Int, Long]())
>
> val buffer = builder.add(Flow[Long].buffer(1,
> OverflowStrategy.dropTail))
>
> unzip.out0 ~> acc ~> zip.in0
> unzip.out1 ~> buffer ~> zip.in1
>
> FlowShape[(Int, Long), (Int, Long)](unzip.in, zip.out)
> }))
>
>
> and I get output, but not what I need:
>
> List((13,0), (11,2), (13,4), (11,7)) was not equal to List((13,0), (11,3),
> (13,5), (11,8))
>
> I think I know what is happening - the first Long element is buffered in
> the zip stage and so the element in the buffer is not actually the one I
> expect. But I don't know how to get around it. Using conflate ends up with
> exactly the same result
>
> Note that I can't easily rewrite the accumulator stage, if I could I would
> just process the pairs together in a stateful stage.
>
> Thanks,
>
> Julian
>
> --
> >>>>>>>>>> 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 https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Cheers,
√
--
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
