title: How do I send message to ActorRef at start of Akka-Stream 2.0 flow?
body: <p>The goal is to send the <code>WSConnectEvent</code> once a client is connected and the stream starts. With akka-streams 1.0 I was able to accomplish this with the following:</p>
  
  <pre><code>Flow(Source.actorRef[WSResponseEvent](65535, OverflowStrategy.fail)) {
    implicit builder =&gt;
      sdpSource =&gt;
  
        // Incoming SDP offer flow
        val fromWebsocket = builder.add(Flow[Message].collect {
          case TextMessage.Strict(txt) =&gt; {
            val event = txt.parseJson.convertTo[WSResponseEvent]
            WSMessageEvent(callUUID, userUUID, event.id, event.data)
          }
        })
  
        // Outgoing SDP answer flow
        val toWebsocket = builder.add(Flow[WSResponseEvent].map {
          case msg: WSResponseEvent =&gt; TextMessage(msg.toJson.compactPrint)
        })
  
        val callActorSelection = actorSystem.actorSelection(s"/user/application/call-service/call-${callUUID.toString}")
        val callActorRef = Await.result(callActorSelection.resolveOne(), Duration.Inf);
        val callActorSink = Sink.actorRef[CallControlEvent](callActorRef, WSDisconnectEvent(callUUID, userUUID))
  
        // Join events, also sends actor for sending stuff
        val merge = builder.add(Merge[CallControlEvent](2))
        val actorAsSource = builder.materializedValue.map(actor =&gt; WSConnectEvent(callUUID, userUUID, actor))
  
        fromWebsocket ~&gt; merge.in(0)
        actorAsSource ~&gt; merge.in(1)
  
        merge ~&gt; callActorSink
        sdpSource ~&gt; toWebsocket
        (fromWebsocket.inlet, toWebsocket.outlet)
  }
  </code></pre>
  
  <p>In trying to upgrade this to work with Akka-Streams 2.0.1 I changed to the following code, but I am no loner receiving the <code>WSConnectEvent</code> message. I'm not sure if this is because my source is setup incorrectly, or I am not materializing the <code>ActorRef</code> properly.</p>
  
  <pre><code>val sdpSource = Source.actorRef[WSResponseEvent](65535, OverflowStrategy.fail)
  
  Flow.fromGraph(
    GraphDSL.create() { implicit builder =&gt;
  
      // Incoming SDP offer flow
      val fromWebsocket = builder.add(Flow[Message].collect {
        case TextMessage.Strict(txt) =&gt; {
          val event = txt.parseJson.convertTo[WSResponseEvent]
          WSMessageEvent(callUUID, userUUID, event.id, event.data)
        }
      })
  
      // Outgoing SDP answer flow
      val toWebsocket = builder.add(Flow[WSResponseEvent].map {
        case msg: WSResponseEvent =&gt; TextMessage(msg.toJson.compactPrint)
      })
  
      val callActorSelection = actorSystem.actorSelection(s"/user/application/call-service/call-${callUUID.toString}")
      val callActorRef = Await.result(callActorSelection.resolveOne(), Duration.Inf);
      val callActorSink = Sink.actorRef[CallControlEvent](callActorRef, WSDisconnectEvent(callUUID, userUUID))
  
      // Join events, also sends actor for sending stuff
      val merge = builder.add(Merge[CallControlEvent](2))
      val actorAsSource = sdpSource.mapMaterializedValue(WSConnectEvent(callUUID, userUUID, _))
  
      fromWebsocket ~&gt; merge.in(0)
      actorAsSource ~&gt; merge.in(1)
  
      merge ~&gt; callActorSink
      sdpSource ~&gt; toWebsocket
      FlowShape(fromWebsocket.in, toWebsocket.out)
    }
  )
  </code></pre>