Flow[Element].grouped(n) solves your problem?
On Tuesday, January 5, 2016 at 6:38:01 PM UTC-2, David Knapp wrote:
>
> I have a stream where a set of N items is parallelized and sent through a
> flow that uses mapAsyncUnordered (an Http().cachedHostConnectionPoolTls[T]
> flow) , and I need the N elements to come out the other side together, but
> not necessarily in order. Basically I need this test to pass:
>
> def httpFlow =
> Flow[(String, Int)]
> .map(s => Post("/", FormData("a" -> s._1)) -> s._2)
> .via(
> Http().cachedHostConnectionPoolTls[Int](
> "akka.io"
> )
> )
> .map(r => r._2)
>
> val testFlow = Flow.fromGraph( GraphDSL.create() { implicit b ⇒
> import GraphDSL.Implicits._
>
> val source = b.add(Broadcast[Int](2))
> val zip = b.add(Zip[Int, Int]())
>
>
> source ~> Flow[Int].map{i => ("Test1", i)} ~> (httpFlow.mapAsync(1){ r =>
> pattern.after(1 second, using = system.scheduler)(Future.successful(r))
> }) ~> zip.in0
> source ~> Flow[Int].map{i => ("Test2", i)} ~> httpFlow ~> zip.in1
>
> FlowShape(source.in, zip.out)
>
> })
>
> val xx = Source(1 to 100)
> .via(testFlow)
> .runWith(TestSink.probe[(Int,Int)])
> .request(100)
>
> xx.within(2 minutes){
> xx.expectNextUnorderedN((1 to 100).map{i => (i, i)})
> }
>
>
> I was thinking of a custom graph stage, but there's not really a default
> shape that seems to fit. I think I need a shape that takes N inputs and has
> N outputs. Is there anything already built anywhere that can do this? Or
> any tips on how to write this?
>
--
>>>>>>>>>> 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.
