feat(remotePlayerEngine): render remote players!

5 files changed, 171 insertions, 65 deletions

diff --git a/domain/src/main/java/akkamon/domain/AkkamonMessageEngine.java b/domain/src/main/java/akkamon/domain/AkkamonMessageEngine.java
index eb998ba..05efc32 100644
--- a/domain/src/main/java/akkamon/domain/AkkamonMessageEngine.java
+++ b/domain/src/main/java/akkamon/domain/AkkamonMessageEngine.java
@@ -4,9 +4,11 @@ import java.util.Map;
 
 public interface AkkamonMessageEngine {
     // broadcasts position info to WebSocket Clients
-    void broadCastHeartBeatToScene(String sceneId, Map<String, AkkamonNexus.TrainerPositionReading> trainerPositions);
+    void broadCastHeartBeatToScene(String sceneId, Map<String, AkkamonNexus.MovementQueueReading> trainerPositions);
 
     void registerTrainerSessionToScene(String sceneId, AkkamonSession session);
 
     void removeTrainerSessionFromScene(String sceneId, AkkamonSession session);
+
+    void trainerDisconnected(AkkamonSession session);
 }
diff --git a/domain/src/main/java/akkamon/domain/AkkamonNexus.java b/domain/src/main/java/akkamon/domain/AkkamonNexus.java
index 9fd2763..a55e728 100644
--- a/domain/src/main/java/akkamon/domain/AkkamonNexus.java
+++ b/domain/src/main/java/akkamon/domain/AkkamonNexus.java
@@ -10,6 +10,7 @@ import akka.actor.typed.javadsl.Receive;
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Queue;
 
 public class AkkamonNexus extends AbstractBehavior<AkkamonNexus.Command> {
 
@@ -126,57 +127,71 @@ public class AkkamonNexus extends AbstractBehavior<AkkamonNexus.Command> {
         }
     }
 
+    public static class RequestTrainerOffline
+            implements Command, SceneTrainerGroup.Command, Trainer.Command {
+        public long requestId;
+        public String trainerId;
+        public String sceneId;
+        public AkkamonSession session;
+        public ActorRef<AkkamonNexus.Command> replyTo;
+
+        public RequestTrainerOffline(long requestId, String trainerId, String sceneId, AkkamonSession session, ActorRef<Command> replyTo) {
+            this.requestId = requestId;
+            this.trainerId = trainerId;
+            this.sceneId = sceneId;
+            this.session = session;
+            this.replyTo = replyTo;
+        }
+    }
+
+    public static class RespondTrainerOffline
+            implements Command {
+        public long requestId;
+        public String sceneId;
+        public AkkamonSession session;
+
+        public RespondTrainerOffline(long requestId, String sceneId, AkkamonSession session) {
+            this.requestId = requestId;
+            this.sceneId = sceneId;
+            this.session = session;
+        }
+    }
+
     public static class RespondHeartBeatQuery implements Command {
 
         public final long requestId;
         public final String sceneId;
-        public final Map<String, TrainerPositionReading> trainerPositions;
+        public final Map<String, MovementQueueReading> trainerMovementQueues;
 
         public RespondHeartBeatQuery(
                 long requestId,
                 String sceneId,
-                Map<String, TrainerPositionReading> trainerPositions) {
+                Map<String, MovementQueueReading> trainerPositions) {
             this.requestId = requestId;
             this.sceneId = sceneId;
-            this.trainerPositions = trainerPositions;
+            this.trainerMovementQueues = trainerPositions;
         }
     }
 
-    public interface TrainerPositionReading { }
+    public interface MovementQueueReading { }
 
-    public static class TrainerPosition implements TrainerPositionReading {
-        public final TilePos value;
+    public static class MovementQueue implements MovementQueueReading {
+        public final Queue<Direction> value;
 
-        public TrainerPosition(TilePos value) {
+        public MovementQueue(Queue<Direction> value) {
             this.value = value;
         }
-
-        @Override
-        public boolean equals(Object o) {
-            if (this == o) return true;
-            if (o == null || getClass() != o.getClass()) return false;
-
-            TrainerPosition other = (TrainerPosition) o;
-
-            return this.value.x == other.value.x && this.value.y == other.value.y;
-        }
-
-        @Override
-        public String toString() {
-            return "TrainerPosition={x: " + value.x + ", y: " + value.y + "}";
-        }
-
     }
 
-    public enum TrainerPositionNotAvailable implements TrainerPositionReading {
+    public enum MovementQueueEmpty implements MovementQueueReading {
         INSTANCE
     }
 
-    public enum TrainerOffline implements TrainerPositionReading {
+    public enum TrainerOffline implements MovementQueueReading {
         INSTANCE
     }
 
-    public enum TrainerTimedOut implements TrainerPositionReading {
+    public enum TrainerTimedOut implements MovementQueueReading {
         INSTANCE
     }
 
@@ -198,6 +213,8 @@ public class AkkamonNexus extends AbstractBehavior<AkkamonNexus.Command> {
         return newReceiveBuilder()
                 .onMessage(RequestTrainerRegistration.class, this::onTrainerRegistration)
                 .onMessage(TrainerRegistered.class, this::onTrainerRegistered)
+                .onMessage(RequestTrainerOffline.class, this::onTrainerOfflineRequest)
+                .onMessage(RespondTrainerOffline.class, this::onTrainerOffline)
                 .onMessage(RequestHeartBeat.class, this::onHeartBeat)
                 .onMessage(RespondHeartBeatQuery.class, this::onHeartBeatQueryResponse)
                 .onMessage(RequestStartMoving.class, this::onStartMoving)
@@ -206,19 +223,37 @@ public class AkkamonNexus extends AbstractBehavior<AkkamonNexus.Command> {
                 .build();
     }
 
+    private AkkamonNexus onTrainerOffline(RespondTrainerOffline trainerOfflineMsg) {
+        getContext().getLog().info("Removing {} from akkamon sessions!", trainerOfflineMsg.session.getTrainerId());
+        messageEngine.removeTrainerSessionFromScene(trainerOfflineMsg.sceneId, trainerOfflineMsg.session);
+        return this;
+    }
+
+    private AkkamonNexus onTrainerOfflineRequest(RequestTrainerOffline trainerOfflineRequest) {
+        ActorRef<SceneTrainerGroup.Command> sceneTrainerGroup = sceneIdToActor.get(
+                trainerOfflineRequest.sceneId
+        );
+        if (sceneTrainerGroup != null) {
+            sceneTrainerGroup.tell(trainerOfflineRequest);
+        } else {
+            getContext().getLog().info("Ignoring trainerOffline request in scene {}, it isn't mapped to a sceneTrainerActor.");
+        }
+        return this;
+    }
+
     private AkkamonNexus onHeartBeatQueryResponse(RespondHeartBeatQuery response) {
         // Turn on for logging
 
 
         StringBuilder positions = new StringBuilder();
         positions.append("\n" + response.sceneId.toUpperCase(Locale.ROOT) + "\n");
-        for (Map.Entry<String, TrainerPositionReading> entry : response.trainerPositions.entrySet()) {
+        for (Map.Entry<String, MovementQueueReading> entry : response.trainerMovementQueues.entrySet()) {
             positions.append(entry.getKey() + ": " +entry.getValue());
             positions.append("\n");
         }
         getContext().getLog().info(String.valueOf(positions));
 
-        messageEngine.broadCastHeartBeatToScene(response.sceneId, response.trainerPositions);
+        messageEngine.broadCastHeartBeatToScene(response.sceneId, response.trainerMovementQueues);
 
        return this;
     }
diff --git a/domain/src/main/java/akkamon/domain/HeartBeatQuery.java b/domain/src/main/java/akkamon/domain/HeartBeatQuery.java
index 4be99ec..5e88ff5 100644
--- a/domain/src/main/java/akkamon/domain/HeartBeatQuery.java
+++ b/domain/src/main/java/akkamon/domain/HeartBeatQuery.java
@@ -5,10 +5,7 @@ import akka.actor.typed.Behavior;
 import akka.actor.typed.javadsl.*;
 
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
 
 public class HeartBeatQuery extends AbstractBehavior<HeartBeatQuery.Command> {
 
@@ -18,11 +15,11 @@ public class HeartBeatQuery extends AbstractBehavior<HeartBeatQuery.Command> {
         INSTANCE
     }
 
-    static class WrappedRespondTrainerPosition implements Command {
-        final Trainer.RespondTrainerPosition response;
+    static class WrappedRespondMovementQueue implements Command {
+        final Trainer.RespondMovementQueue response;
 
-        WrappedRespondTrainerPosition(
-                Trainer.RespondTrainerPosition response
+        WrappedRespondMovementQueue(
+                Trainer.RespondMovementQueue response
         ) {
             this.response = response;
         }
@@ -62,7 +59,7 @@ public class HeartBeatQuery extends AbstractBehavior<HeartBeatQuery.Command> {
     private final long requestId;
     private final String sceneId;
     private final ActorRef<AkkamonNexus.Command> requester;
-    private Map<String, AkkamonNexus.TrainerPositionReading> repliesSoFar = new HashMap<String, AkkamonNexus.TrainerPositionReading>();
+    private Map<String, AkkamonNexus.MovementQueueReading> repliesSoFar = new HashMap<String, AkkamonNexus.MovementQueueReading>();
     private final Set<String> stillWaiting;
 
     public HeartBeatQuery(
@@ -80,13 +77,13 @@ public class HeartBeatQuery extends AbstractBehavior<HeartBeatQuery.Command> {
 
         timers.startSingleTimer(CollectionTimeout.INSTANCE, timeout);
 
-        ActorRef<Trainer.RespondTrainerPosition> respondTrainerPositionAdapter =
-                context.messageAdapter(Trainer.RespondTrainerPosition.class, WrappedRespondTrainerPosition::new);
+        ActorRef<Trainer.RespondMovementQueue> respondTrainerPositionAdapter =
+                context.messageAdapter(Trainer.RespondMovementQueue.class, WrappedRespondMovementQueue::new);
 
         for (Map.Entry<String, ActorRef<Trainer.Command>> entry : trainerIdToActor.entrySet()) {
             context.watchWith(entry.getValue(), new TrainerOffline(entry.getKey()));
             entry.getValue().tell(
-                    new Trainer.ReadTrainerPosition(
+                    new Trainer.ReadMovementQueue(
                             0L,
                             respondTrainerPositionAdapter
                     )
@@ -98,19 +95,22 @@ public class HeartBeatQuery extends AbstractBehavior<HeartBeatQuery.Command> {
     @Override
     public Receive<Command> createReceive() {
         return newReceiveBuilder()
-                .onMessage(WrappedRespondTrainerPosition.class, this::onRespondTrainerPosition)
+                .onMessage(WrappedRespondMovementQueue.class, this::onRespondMovementQueue)
                 .build();
     }
 
-    private Behavior<Command> onRespondTrainerPosition(WrappedRespondTrainerPosition r) {
-        AkkamonNexus.TrainerPositionReading trainerPositionRead =
-                r.response
-                .value
-                .map(optionalValue -> (AkkamonNexus.TrainerPositionReading) new AkkamonNexus.TrainerPosition(optionalValue))
-                .orElse(AkkamonNexus.TrainerPositionNotAvailable.INSTANCE);
+    private Behavior<Command> onRespondMovementQueue(WrappedRespondMovementQueue r) {
+        AkkamonNexus.MovementQueueReading movementQueueRead = null;
+        if (r.response.value.size() != 0) {
+            movementQueueRead = new AkkamonNexus.MovementQueue(r.response.value);
+        } else {
+            Queue<Direction> queue = new LinkedList<>();
+            queue.add(Direction.NONE);
+            movementQueueRead = new AkkamonNexus.MovementQueue(queue);
+        }
 
         String trainerId = r.response.trainerId;
-        repliesSoFar.put(trainerId, trainerPositionRead);
+        repliesSoFar.put(trainerId, movementQueueRead);
         stillWaiting.remove(trainerId);
 
         return respondWhenAllCollected();
diff --git a/domain/src/main/java/akkamon/domain/SceneTrainerGroup.java b/domain/src/main/java/akkamon/domain/SceneTrainerGroup.java
index 29778ec..e8ace82 100644
--- a/domain/src/main/java/akkamon/domain/SceneTrainerGroup.java
+++ b/domain/src/main/java/akkamon/domain/SceneTrainerGroup.java
@@ -15,15 +15,18 @@ public class SceneTrainerGroup extends AbstractBehavior<SceneTrainerGroup.Comman
 
     public interface Command { }
 
-    private class TrainerOffline implements Command {
+    public static class TrainerOffline
+            implements Command, AkkamonNexus.Command {
         public ActorRef<Trainer.Command> trainer;
         public String sceneId;
         public String trainerId;
+        public ActorRef<AkkamonNexus.Command> replyTo;
 
-        public TrainerOffline(ActorRef<Trainer.Command> trainerActor, String sceneId, String trainerId) {
+        public TrainerOffline(ActorRef<Trainer.Command> trainerActor, String sceneId, String trainerId, ActorRef<AkkamonNexus.Command> replyTo) {
             this.trainer = trainerActor;
             this.sceneId = sceneId;
             this.trainerId = trainerId;
+            this.replyTo = replyTo;
         }
     }
 
@@ -50,6 +53,14 @@ public class SceneTrainerGroup extends AbstractBehavior<SceneTrainerGroup.Comman
                         this::onTrainerRegistration
                 )
                 .onMessage(
+                        AkkamonNexus.RequestTrainerOffline.class,
+                        this::onTrainerOfflineRequest
+                )
+                .onMessage(
+                        TrainerOffline.class,
+                        this::onWatchedTrainerOffline
+                )
+                .onMessage(
                         AkkamonNexus.RequestStartMoving.class,
                         this::onStartMoving
                 )
@@ -68,6 +79,41 @@ public class SceneTrainerGroup extends AbstractBehavior<SceneTrainerGroup.Comman
                 .build();
     }
 
+    private SceneTrainerGroup onWatchedTrainerOffline(TrainerOffline trainerOfflineMsg) {
+        trainerOfflineMsg.replyTo.tell(trainerOfflineMsg);
+        trainerIdToActor.remove(trainerOfflineMsg.trainerId);
+        return this;
+    }
+
+    private SceneTrainerGroup onTrainerOfflineRequest(AkkamonNexus.RequestTrainerOffline trainerOfflineRequest) {
+        if (this.sceneId.equals(trainerOfflineRequest.sceneId)) {
+            ActorRef<Trainer.Command> trainerActor = trainerIdToActor.get(trainerOfflineRequest.trainerId);
+            if (trainerActor != null) {
+                trainerActor.tell(trainerOfflineRequest);
+                trainerOfflineRequest.replyTo.tell(new AkkamonNexus.RespondTrainerOffline(
+                        trainerOfflineRequest.requestId,
+                        trainerOfflineRequest.sceneId,
+                        trainerOfflineRequest.session
+                ));
+            } else {
+                getContext()
+                        .getLog()
+                        .warn(
+                                "Ignoring trainerOffline for trainerId {}. There is no actor mapped to it.",
+                                trainerOfflineRequest.trainerId
+                        );
+            }
+        } else {
+            getContext()
+                    .getLog()
+                    .warn(
+                            "Ignoring trainerOffline for {}. This actor is responsible for {}.",
+                            trainerOfflineRequest.sceneId,
+                            this.sceneId);
+        }
+        return this;
+    }
+
     private SceneTrainerGroup onHeartBeat(AkkamonNexus.RequestHeartBeat heartBeatRequest) {
         Map<String, ActorRef<Trainer.Command>> trainerIdToActorCopy = new HashMap<>(this.trainerIdToActor);
         getContext()
@@ -171,7 +217,7 @@ public class SceneTrainerGroup extends AbstractBehavior<SceneTrainerGroup.Comman
                         getContext()
                                 .spawn(Trainer.create(sceneId, registrationRequest.trainerId), "trainer-" + registrationRequest.trainerId);
                 getContext()
-                        .watchWith(trainerActor, new SceneTrainerGroup.TrainerOffline(trainerActor, sceneId, registrationRequest.trainerId));
+                        .watchWith(trainerActor, new SceneTrainerGroup.TrainerOffline(trainerActor, sceneId, registrationRequest.trainerId, registrationRequest.replyTo));
                 trainerIdToActor.put(registrationRequest.trainerId, trainerActor);
                 registrationRequest.replyTo.tell(new AkkamonNexus.TrainerRegistered(
                         registrationRequest.trainerId,
diff --git a/domain/src/main/java/akkamon/domain/Trainer.java b/domain/src/main/java/akkamon/domain/Trainer.java
index 0a3b397..710f844 100644
--- a/domain/src/main/java/akkamon/domain/Trainer.java
+++ b/domain/src/main/java/akkamon/domain/Trainer.java
@@ -7,31 +7,33 @@ import akka.actor.typed.javadsl.ActorContext;
 import akka.actor.typed.javadsl.Behaviors;
 import akka.actor.typed.javadsl.Receive;
 
+import java.util.LinkedList;
 import java.util.Optional;
+import java.util.Queue;
 
 public class Trainer extends AbstractBehavior<Trainer.Command> {
 
     public interface Command { }
 
-    public static class ReadTrainerPosition implements Command {
+    public static class ReadMovementQueue implements Command {
         final long requestId;
-        final ActorRef<RespondTrainerPosition> replyTo;
+        final ActorRef<RespondMovementQueue> replyTo;
 
-        public ReadTrainerPosition(long requestId, ActorRef<RespondTrainerPosition> replyTo) {
+        public ReadMovementQueue(long requestId, ActorRef<RespondMovementQueue> replyTo) {
             this.requestId = requestId;
             this.replyTo = replyTo;
         }
     }
 
-    public static final class RespondTrainerPosition {
+    public static final class RespondMovementQueue {
         final long requestId;
         final String trainerId;
-        final Optional<TilePos> value;
+        final Queue<Direction> value;
 
-        public RespondTrainerPosition(
+        public RespondMovementQueue(
                 long requestId,
                 String trainerId,
-                Optional<TilePos> value
+                Queue<Direction> value
         ) {
             this.requestId = requestId;
             this.trainerId = trainerId;
@@ -46,6 +48,10 @@ public class Trainer extends AbstractBehavior<Trainer.Command> {
     private String sceneId;
     private String trainerId;
 
+    private Queue<Direction> movementQueue = new LinkedList<>();
+
+    private Direction movementDirection = Direction.NONE;
+
     private Optional<TilePos> lastValidTilePos = Optional.empty();
 
     public Trainer(ActorContext<Command> context, String sceneId, String trainerId) {
@@ -58,8 +64,12 @@ public class Trainer extends AbstractBehavior<Trainer.Command> {
     public Receive<Command> createReceive() {
         return newReceiveBuilder()
                 .onMessage(
-                        ReadTrainerPosition.class,
-                        this::onReadTrainerPosition
+                        ReadMovementQueue.class,
+                        this::onReadMovementQueue
+                )
+                .onMessage(
+                        AkkamonNexus.RequestTrainerOffline.class,
+                        this::onTrainerOffline
                 )
                 .onMessage(
                         AkkamonNexus.RequestStartMoving.class,
@@ -75,29 +85,42 @@ public class Trainer extends AbstractBehavior<Trainer.Command> {
                 .build();
     }
 
-    private Trainer onReadTrainerPosition(ReadTrainerPosition readTrainerPositionRequest) {
-        readTrainerPositionRequest.replyTo.tell(new RespondTrainerPosition(
+    private Behavior<Command> onTrainerOffline(AkkamonNexus.RequestTrainerOffline trainerOfflineRequest) {
+        getContext().getLog().info("Trainer {} went offline, the actor has stopped! My supervisor should handle closing my connection!");
+        return Behaviors.stopped();
+    }
+
+    private Trainer onReadMovementQueue(ReadMovementQueue readTrainerPositionRequest) {
+        readTrainerPositionRequest.replyTo.tell(new RespondMovementQueue(
                 readTrainerPositionRequest.requestId,
                 trainerId,
-                lastValidTilePos
+                new LinkedList<>(movementQueue)
         ));
+        this.movementQueue.clear();
         return this;
     }
 
     private Trainer onNewTilePos(AkkamonNexus.RequestNewTilePos newTilePosRequest) {
         getContext().getLog().info("Trainer {} has new {}.", trainerId, newTilePosRequest.tilePos);
-        lastValidTilePos = Optional.of(newTilePosRequest.tilePos);
+        if (isMoving()) {
+            this.movementQueue.add(this.movementDirection);
+        }
         return this;
     }
 
     private Trainer onStopMoving(AkkamonNexus.RequestStopMoving stopMovingRequest) {
         getContext().getLog().info("Trainer {} stops to move {}.", trainerId, stopMovingRequest.direction);
+        this.movementDirection = Direction.NONE;
         return this;
     }
 
     private Trainer onStartMoving(AkkamonNexus.RequestStartMoving startMovingRequest) {
         getContext().getLog().info("Trainer {} starts to move {}.", trainerId, startMovingRequest.direction);
+        this.movementDirection = startMovingRequest.direction;
         return this;
     }
 
+    private boolean isMoving() {
+        return this.movementDirection != Direction.NONE;
+    }
 }