akgupta

getContext().watch(_eventSourceActor);

@Override
public Receive createReceive() {
  return receiveBuilder()
    .match(
      Terminated.class,
      terminated -> {
        if (terminated.getActor().equals(_eventSourceActor)) {
            getContext().stop(self());

akgupta

_eventSourceActor.tell(new Status.Success("close"), getSelf())

akgupta

_eventSourceActor.tell(event(Json.toJson(new ClientMessage(_connectionId, "heartbeat"))), getSelf())

akgupta

return ok().chunked(eventSource.via(EventSource.flow())).as(Http.MimeTypes.EVENT_STREAM);

akgupta

String connectionIdentifier = UUID.randomUUID().toString();
_actorSystem.actorOf(
    ClientConnection.getProps(
      connectionIdentifier, eventSourceActor, _actorSystem.getScheduler()
    ),
    connectionIdentifier
);

akgupta

Pair<ActorRef, Source<EventSource.Event, NotUsed>> actorRefEventSourcePair =
  actorRefPoweredEventSource.preMaterialize(_materializer);
ActorRef eventSourceActor = actorRefEventSourcePair.first();
Source<EventSource.Event, ?> eventSource = actorRefEventSourcePair.second();

akgupta

Source<EventSource.Event, ActorRef> actorRefPoweredEventSource =
  Source.actorRef(100, OverflowStrategy.fail());

akgupta

"Hashed wheel timer #11327" #27780 prio=5 os_prio=0 tid=0x00007f73a8bba000 nid=0x27f4 sleeping[0x00007f7329d23000]
   java.lang.Thread.State: TIMED_WAITING (sleeping)
    at java.lang.Thread.sleep(Native Method)
    at org.jboss.netty.util.HashedWheelTimer$Worker.waitForNextTick(HashedWheelTimer.java:445)
    at org.jboss.netty.util.HashedWheelTimer$Worker.run(HashedWheelTimer.java:364)
    at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
    at java.lang.Thread.run(Thread.java:745)

akgupta

public class ClientConnectionActor extends UntypedActor {
  public static Props props(String connectionId, EventSource eventSource) {
    return Props.create(ClientConnectionActor.class, () -> new ClientConnectionActor(connectionId, eventSource));
  }
  
  public void onReceive(Object msg) throws Exception {
    if (msg instanceof ClientMessage) {
      eventSource.send(event(Json.toJson(clientMessage)));
    }
  }

akgupta

// User B sends a message to User A
// We identify the Actor which manages the connection on which User A is connected (connectionIdA)
ActorSelection actorSelection = Akka.system().actorSelection("akka://application/user/" + connectionIdA);
// Send B's message to A's Actor
actorSelection.tell(new ClientMessage(data), ActorRef.noSender());