hi , i got a faster upstream, and a slow downstream,
so i want to "combine" the up-coming item into a arraybuffer, and then
consume the arraybuffer in a batch.
i find the Flow.batch seems can help me here,
but i write some test,which is not what i expect
*case class *IntHolder(*var *a:Int=0)
test(*"test simple batch"*) {
*implicit val *mat = *ActorMaterializer*()
*val *broadCastModule = *new *BroadCastModule[String](*system*)
*val *source = Source.*repeat*(1)
*// val source = Source.actorRef[Int](1000, OverflowStrategy.dropHead)
**val *temp=source.batch(20,it=>*new
*IntHolder(0))((all,other)=>{all.a+=other;all})
*val *end=temp.runForeach(it=>{*println*(it);Thread.*sleep*(2000)})
Thread.*sleep*(50000)
}
most output is:
IntHolder(1)
IntHolder(1)
IntHolder(1)
IntHolder(1)
IntHolder(1)
here is my code, the problem is the most of IntHolder is 1 inside
but i thread.sleep(2000) so i think the consumer is slow enough,
i expect most of the println is IntHolder(20),because the consumer is slow,the
producer will do batch for most item.
what is wrong? thanks
--
>>>>>>>>>> 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.
