I think one of the keys here is that the materialized value (the
Future[Long] in this case) is not an element, per se, so it won't get
streamed inline.
In other words, don't expect to do a broadcast/zip to get the materialized
values.
You'll need to capture the materialized value when the stream is
materialized, not when the flow is constructed.
I don't know your exact logic, but here's a quick stab at your problem in
the form of a ScalaTest. I'm creating the sink as a separate step and I
added some verbose typing, etc. to illustrate and hopefully make things
clear. Of course, imports and such omitted for brevity.
Note the use of the materialized value combiner function (m1, m2) => (m1,
m2). This will get you both materialized values out of the sink.
def outputSink(outStream1: OutputStream, outStream2: OutputStream):
Sink[ByteString, (Future[Long], Future[Long])] = {
import akka.stream.scaladsl.FlowGraph.Implicits._
val outSink1 = OutputStreamSink(() => outStream1)
val outSink2 = OutputStreamSink(() => outStream2)
Sink(outSink1, outSink2)((m1, m2) => (m1, m2)) { implicit builder =>
(out1, out2) => {
val bcast = builder.add(Broadcast[ByteString](2))
bcast ~> out1
bcast ~> out2
(bcast.in)
}
}
}
"Flow" should "produce materialized values" in {
val data = "Some String"
val source = InputStreamSource(() => new
ByteArrayInputStream(data.getBytes))
val outStream1 = new ByteArrayOutputStream()
val outStream2 = new ByteArrayOutputStream()
val sink = outputSink(outStream1, outStream2)
val runnable: RunnableFlow[(Future[Long], Future[Long])] =
source.toMat(sink)(Keep.right)
val (size1, size2) = runnable.run()
whenReady(size1) { size =>
size shouldBe data.getBytes.length
}
whenReady(size2) { size =>
size shouldBe data.getBytes.length
}
Regards,
Lance
On Saturday, June 20, 2015 at 8:13:08 AM UTC-4, Michael Hamrah wrote:
>
> I'm working with an OutputStreamSink (
> http://doc.akka.io/api/akka-stream-and-http-experimental/1.0-RC3/index.html#akka.stream.io.OutputStreamSink$)
>
> which materializes a Future[Long]. I'd like to use an OutputStreamSink as
> part of a flow, but I can't get the types to line up.
>
> Ideally my graph will look something like this; it splits an input stream,
> zips the result, and checker checks to make sure the future[long]'s are
> equal.
>
> ```
> in ~> broadcast ~> outputsink1 ~> zip.in0 ~> checker
> ~> broadcast ~> outputsink2 ~> zip.in1
> ```
>
> any ideas? It seems like I need to rewrite OutputStreamSink from a Sink to
> a Flow, or use a SubscriberActor with a PublisherActor, or possibly write
> some sort of PushPullStage. Seems like there should be an easy way to
> "lift" a sink with a materialized value to a flow stage.
>
> Thanks,
>
> Mike
>
--
>>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
