Hi Jeff,
Since what you want to do is materialize a stream on each such binary
message,
the case block does not happen upon creation of your Flow or even upon
materialization
of it, you will need access to the materializer inside of it.
I don't think you have any other option than to make materializer available
in your apply
method, from a method parameter for example.
Side note: I see you are still on an old version of akka streams, please
try to
upgrade to 2.0.x as soon as you can, it contains many fixes and
improvements!
--
Johan Andrén
Typesafe - Reactive apps on the JVM
Twitter: @apnylle
On Monday, December 14, 2015 at 10:54:18 PM UTC+1, Jeff wrote:
>
> I am working my way through the websocket example and have a question
> about nested streams. Currently I have this:
>
> import akka.actor.ActorRef
> import akka.http.scaladsl.model.ws.{BinaryMessage, TextMessage, Message}
> import akka.stream.{OverflowStrategy, FlowShape}
> import akka.stream.scaladsl.{Sink, Source, FlowGraph, Flow}
>
> object ChatFlow {
> def apply(userId: Long, channel: Long, rabbitActor: ActorRef):
> Flow[Message, Message, ActorRef] =
> Flow.fromGraph(FlowGraph.create(Source.actorRef[RabbitActor.Message](5,
> OverflowStrategy.fail)) { implicit b =>
> rabbitSource =>
> import FlowGraph.Implicits._
>
> // for now, ignore all content coming from the websocket
> val fromWebsocket = b.add(Sink.foreach[Message] {
> case TextMessage.Strict(content) => println(content)
> // case bm: BinaryMessage => bm.dataStream.runWith(Sink.ignore)
> })
>
> // translate rabbit messages into socket messages
> val toWebSocket = b.add(Flow[RabbitActor.Message] map {
> case _ => TextMessage(s"test")
> })
>
> // when this stream is materialized, register it with rabbitmq
> val websocketMaterialized = b.materializedValue map { actor =>
> RabbitActor.UserJoined(userId, channel, actor)
> }
>
> val rabbitMQSink = Sink.actorRef[RabbitActor.Message](rabbitActor,
> RabbitActor.UserLeft(userId, channel))
>
> websocketMaterialized ~> rabbitMQSink
> rabbitSource ~> toWebSocket
>
> FlowShape(fromWebsocket.inlet, toWebSocket.outlet)
> })
> }
>
>
> My question is about the commented out BinaryMessage. bm.dataStream is a
> Source, and the docs say I should drain this stream to clear the socket.
> How would I do that in this scenario without passing in a materializer?
>
--
>>>>>>>>>> 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 https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
