title: How to use an Actor based source with an Akka Graph?
body: <p>I'm attempting to send data via an actor to a runnable graph that contains a fan out.</p>
  <p>I define the source as :</p>
  <pre><code>final Source&lt;Integer, ActorRef&gt; integerSource =
          Source.actorRef(
                  elem -&gt; {
                      if (elem == Done.done()) return Optional.of(CompletionStrategy.immediately());
                      else return Optional.empty();
                  },
                  elem -&gt; Optional.empty(),
                  10,
                  OverflowStrategy.dropHead());
  </code></pre>
  <p>But I'm unsure how to get a handle on an <code>ActoRef</code> to send data via an actor to the source so that the runnable graph will process messages asynchronously as they are received :</p>
  <pre><code>RunnableGraph&lt;CompletionStage&lt;Done&gt;&gt; graph = RunnableGraph.fromGraph(
          GraphDSL.create(sink, (builder, out) -&gt; {
              SourceShape&lt;Integer&gt; sourceShape = builder.add(integerSource);
              FlowShape&lt;Integer, Integer&gt; flow1Shape = builder.add(flow1);
              FlowShape&lt;Integer, Integer&gt; flow2Shape = builder.add(flow1);
              UniformFanOutShape&lt;Integer, Integer&gt; broadcast =
                      builder.add(Broadcast.create(2));
              UniformFanInShape&lt;Integer, Integer&gt; merge =
                      builder.add(Merge.create(2));
  
              builder.from(sourceShape)
                      .viaFanOut(broadcast)
                      .via(flow1Shape);
  
              builder.from(broadcast).via(flow2Shape);
  
              builder.from(flow1Shape)
                      .viaFanIn(merge)
                      .to(out);
  
              builder.from(flow2Shape).viaFanIn(merge);
  
  
              return ClosedShape.getInstance();
          } )
  );
  </code></pre>
  <p>Entire src :</p>
  <pre><code>import akka.Done;
  import akka.NotUsed;
  import akka.actor.ActorRef;
  import akka.actor.typed.ActorSystem;
  import akka.actor.typed.javadsl.Behaviors;
  import akka.stream.*;
  import akka.stream.javadsl.*;
  import lombok.extern.slf4j.Slf4j;
  
  import java.util.Optional;
  import java.util.concurrent.CompletionStage;
  
  @Slf4j
  public class GraphActorSource {
  
      private final static ActorSystem actorSystem = ActorSystem.create(Behaviors.empty(), &quot;flowActorSystem&quot;);
  
      public void runFlow() {
  
  
          final Source&lt;Integer, ActorRef&gt; integerSource =
                  Source.actorRef(
                          elem -&gt; {
                              if (elem == Done.done()) return Optional.of(CompletionStrategy.immediately());
                              else return Optional.empty();
                          },
                          elem -&gt; Optional.empty(),
                          10,
                          OverflowStrategy.dropHead());
  
  
          Flow&lt;Integer, Integer, NotUsed&gt; flow1 = Flow.of(Integer.class)
                  .map (x -&gt; {
                      System.out.println(&quot;Flow 1 is processing &quot; + x);
                      return (x * 2);
                  });
  
          Sink&lt;Integer, CompletionStage&lt;Done&gt;&gt; sink = Sink.foreach(x -&gt; {
              System.out.println(x);
          });
  
          RunnableGraph&lt;CompletionStage&lt;Done&gt;&gt; graph = RunnableGraph.fromGraph(
                  GraphDSL.create(sink, (builder, out) -&gt; {
                      SourceShape&lt;Integer&gt; sourceShape = builder.add(integerSource);
                      FlowShape&lt;Integer, Integer&gt; flow1Shape = builder.add(flow1);
                      FlowShape&lt;Integer, Integer&gt; flow2Shape = builder.add(flow1);
                      UniformFanOutShape&lt;Integer, Integer&gt; broadcast =
                              builder.add(Broadcast.create(2));
                      UniformFanInShape&lt;Integer, Integer&gt; merge =
                              builder.add(Merge.create(2));
  
                      builder.from(sourceShape)
                              .viaFanOut(broadcast)
                              .via(flow1Shape);
  
                      builder.from(broadcast).via(flow2Shape);
  
                      builder.from(flow1Shape)
                              .viaFanIn(merge)
                              .to(out);
  
                      builder.from(flow2Shape).viaFanIn(merge);
  
  
                      return ClosedShape.getInstance();
                  } )
          );
  
          graph.run(actorSystem);
  
      }
  
      public static void main(String args[]){
          new GraphActorSource().runFlow();
      }
  }
  </code></pre>
  <p>How to send data to the Runnable graph via an actor?</p>
  <p>Something like ? :</p>
  <pre><code>integerSource.tell(1)
  integerSource.tell(2)
  integerSource.tell(3)
  </code></pre>