working ws connection & player updates

Author: naturefreshmilk

src/main/java/io/rudin/minetest/tileserver/TileServer.java

@@ -1,11 +1,5 @@
 package io.rudin.minetest.tileserver;
 
-import static spark.Spark.get;
-import static spark.Spark.init;
-import static spark.Spark.port;
-import static spark.Spark.staticFileLocation;
-import static spark.Spark.stop;
-
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -15,43 +9,52 @@
 
 import io.rudin.minetest.tileserver.config.TileServerConfig;
 import io.rudin.minetest.tileserver.job.UpdateChangedTilesJob;
+import io.rudin.minetest.tileserver.job.UpdatePlayerJob;
 import io.rudin.minetest.tileserver.module.ConfigModule;
 import io.rudin.minetest.tileserver.module.DBModule;
 import io.rudin.minetest.tileserver.module.ServiceModule;
 import io.rudin.minetest.tileserver.route.PlayerRoute;
 import io.rudin.minetest.tileserver.route.TileRoute;
 import io.rudin.minetest.tileserver.transformer.JsonTransformer;
+import io.rudin.minetest.tileserver.ws.WebSocketHandler;
+import io.rudin.minetest.tileserver.ws.WebSocketUpdater;
+
+import static spark.Spark.*;
 
 public class TileServer {
 
-	private static Injector injector;
+	private static Injector injector = Guice.createInjector(
+			new ConfigModule(),
+			new DBModule(),
+			new ServiceModule()
+	);
 
 	public static void main(String[] args) throws Exception {
-		
-		Injector injector = Guice.createInjector(
-				new ConfigModule(),
-				new DBModule(),
-				new ServiceModule()
-		);
-		
+
 		TileServerConfig cfg = injector.getInstance(TileServerConfig.class);
 
 		DBMigration dbMigration = injector.getInstance(DBMigration.class);
 		dbMigration.migrate();
 
 		staticFileLocation("/public");
 		port(cfg.httPort());
-		init();
-		
+
 		JsonTransformer json = injector.getInstance(JsonTransformer.class);
-		
+
+		webSocket("/ws", WebSocketHandler.class);
 		get("/tiles/:z/:x/:y", injector.getInstance(TileRoute.class));
 		get("/player", injector.getInstance(PlayerRoute.class), json);
-		
+
+		//Initialize web server
+		init();
+
+		//Initialize ws updater
+		injector.getInstance(WebSocketUpdater.class).init();
+
 		ScheduledExecutorService executor = injector.getInstance(ScheduledExecutorService.class);
-		UpdateChangedTilesJob tilesJob = injector.getInstance(UpdateChangedTilesJob.class);
-		
-		executor.scheduleAtFixedRate(tilesJob, 0, 2, TimeUnit.SECONDS);
+
+		executor.scheduleAtFixedRate(injector.getInstance(UpdateChangedTilesJob.class), 0, 2, TimeUnit.SECONDS);
+		executor.scheduleAtFixedRate(injector.getInstance(UpdatePlayerJob.class), 0, 1, TimeUnit.SECONDS);
 
 		AtomicBoolean running = new AtomicBoolean(true);
 

src/main/java/io/rudin/minetest/tileserver/job/UpdateChangedTilesJob.java

@@ -5,6 +5,7 @@
 import javax.inject.Inject;
 import javax.inject.Singleton;
 
+import io.rudin.minetest.tileserver.service.EventBus;
 import org.jooq.Cursor;
 import org.jooq.DSLContext;
 import org.jooq.Record2;
@@ -17,11 +18,14 @@
 public class UpdateChangedTilesJob implements Runnable {
 
 	@Inject
-	public UpdateChangedTilesJob(DSLContext ctx, TileCache tileCache) {
+	public UpdateChangedTilesJob(DSLContext ctx, TileCache tileCache, EventBus eventBus) {
 		this.ctx = ctx;
 		this.tileCache = tileCache;
+		this.eventBus = eventBus;
 	}
 
+	private final EventBus eventBus;
+
 	private final DSLContext ctx;
 
 	private final TileCache tileCache;
@@ -52,19 +56,25 @@ public void run() {
 					Integer x = change.get(TILESERVER_BLOCK_CHANGES.POSX);
 					Integer z = change.get(TILESERVER_BLOCK_CHANGES.POSZ);
 
-					System.out.println("Mapblock changed: " + x + "/" + z + "  (Coordinates: " + x*16 + "/" + z*16 + ")");
 
 					TileInfo tileInfo = CoordinateResolver.fromCoordinatesMinZoom(x, z);
 
 					//remove all tiles in every zoom
 					for (int i=CoordinateResolver.MAX_ZOOM; i>=CoordinateResolver.MIN_ZOOM; i--) {
 						TileInfo zoomedTile = tileInfo.toZoom(i);
-						
+
+						EventBus.TileChangedEvent event = new EventBus.TileChangedEvent();
+						event.x = zoomedTile.x;
+						event.y = zoomedTile.y;
+						event.zoom = zoomedTile.zoom;
+						event.mapblockX = x;
+						event.mapblockZ = z;
+						eventBus.post(event);
+
 						tileCache.remove(zoomedTile.x, zoomedTile.y, zoomedTile.zoom);
 					}
-					
-					//TODO: event-bus for ui notification
-					
+
+					//TODO: atomic change
 					ctx
 					.update(TILESERVER_BLOCK_CHANGES)
 					.set(TILESERVER_BLOCK_CHANGES.CHANGED, false)

src/main/java/io/rudin/minetest/tileserver/job/UpdatePlayerJob.java

@@ -0,0 +1,65 @@
+package io.rudin.minetest.tileserver.job;
+
+import io.rudin.minetest.tileserver.blockdb.tables.pojos.Player;
+import io.rudin.minetest.tileserver.service.EventBus;
+import org.jooq.DSLContext;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import java.sql.Timestamp;
+import java.util.List;
+
+import static io.rudin.minetest.tileserver.blockdb.tables.Player.PLAYER;
+
+@Singleton
+public class UpdatePlayerJob implements Runnable {
+
+    @Inject
+    public UpdatePlayerJob(DSLContext ctx, EventBus eventBus){
+        this.ctx = ctx;
+        this.eventBus = eventBus;
+    }
+
+    private final DSLContext ctx;
+
+    private final EventBus eventBus;
+
+    private Timestamp timestamp = new Timestamp(0L);
+
+    private boolean running = false;
+
+    @Override
+    public void run() {
+        if (running)
+            return;
+
+        try {
+            running = true;
+
+            List<Player> players = ctx
+                    .selectFrom(PLAYER)
+                    .where(PLAYER.MODIFICATION_DATE.gt(timestamp))
+                    .fetch()
+                    .into(Player.class);
+
+            for (Player player : players) {
+
+                Timestamp modificationDate = player.getModificationDate();
+
+                if (modificationDate.after(timestamp)) {
+                    //Rember newest modification date
+                    this.timestamp = modificationDate;
+                }
+
+                EventBus.PlayerMovedEvent event = new EventBus.PlayerMovedEvent();
+                event.player = player;
+                eventBus.post(event);
+            }
+        } finally {
+            running = false;
+
+        }
+
+    }
+}

src/main/java/io/rudin/minetest/tileserver/module/ServiceModule.java

@@ -8,15 +8,18 @@
 import io.rudin.minetest.tileserver.ColorTable;
 import io.rudin.minetest.tileserver.provider.ColorTableProvider;
 import io.rudin.minetest.tileserver.provider.ExecutorProvider;
+import io.rudin.minetest.tileserver.service.EventBus;
 import io.rudin.minetest.tileserver.service.TileCache;
 import io.rudin.minetest.tileserver.service.impl.DatabaseTileCache;
+import io.rudin.minetest.tileserver.service.impl.EventBusImpl;
 
 public class ServiceModule extends AbstractModule {
 
 	@Override
 	protected void configure() {
 		bind(TileCache.class).to(DatabaseTileCache.class);
 		//bind(TileCache.class).to(FileTileCache.class);
+		bind(EventBus.class).to(EventBusImpl.class);
 		bind(ColorTable.class).toProvider(ColorTableProvider.class);
 		bind(ExecutorService.class).toProvider(ExecutorProvider.class);
 		bind(ScheduledExecutorService.class).toProvider(ExecutorProvider.class);

src/main/java/io/rudin/minetest/tileserver/service/EventBus.java

@@ -0,0 +1,20 @@
+package io.rudin.minetest.tileserver.service;
+
+import io.rudin.minetest.tileserver.blockdb.tables.pojos.Player;
+
+public interface EventBus {
+
+    void post(Object obj);
+
+    void register(Object listener);
+
+    class TileChangedEvent {
+        public int x,y,zoom;
+        public int mapblockX, mapblockZ;
+    }
+
+    class PlayerMovedEvent {
+        public Player player;
+    }
+
+}

src/main/java/io/rudin/minetest/tileserver/service/impl/EventBusImpl.java

@@ -0,0 +1,34 @@
+package io.rudin.minetest.tileserver.service.impl;
+
+import io.rudin.minetest.tileserver.TileServer;
+import io.rudin.minetest.tileserver.blockdb.tables.Player;
+import io.rudin.minetest.tileserver.service.EventBus;
+
+import javax.inject.Singleton;
+
+@Singleton
+public class EventBusImpl implements EventBus {
+
+    private final com.google.common.eventbus.EventBus eventBus = new com.google.common.eventbus.EventBus();
+
+    @Override
+    public void post(Object obj) {
+
+        if (obj instanceof PlayerMovedEvent){
+            PlayerMovedEvent e = (PlayerMovedEvent)obj;
+            System.out.println("Player-move: " + e.player.getName() + " @" + e.player.getPosx() + "/" + e.player.getPosz());
+        }
+
+        if (obj instanceof TileChangedEvent){
+            //TileChangedEvent e = (TileChangedEvent)obj;
+            //System.out.println("Mapblock changed: " + e.mapblockX + "/" + e.mapblockZ + "  (Coordinates: " + e.mapblockX*16 + "/" + e.mapblockZ*16 + ") @ zoom " + e.zoom);
+        }
+
+        eventBus.post(obj);
+    }
+
+    @Override
+    public void register(Object listener) {
+        eventBus.register(listener);
+    }
+}

src/main/java/io/rudin/minetest/tileserver/ws/WebSocketHandler.java

@@ -0,0 +1,28 @@
+package io.rudin.minetest.tileserver.ws;
+
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+
+import java.io.IOException;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+@WebSocket
+public class WebSocketHandler {
+
+    public static final Queue<Session> sessions = new ConcurrentLinkedQueue<>();
+
+    @OnWebSocketConnect
+    public void connected(Session session) {
+        sessions.add(session);
+    }
+
+    @OnWebSocketClose
+    public void closed(Session session, int statusCode, String reason) {
+        sessions.remove(session);
+    }
+
+}

src/main/java/io/rudin/minetest/tileserver/ws/WebSocketUpdater.java

@@ -0,0 +1,75 @@
+package io.rudin.minetest.tileserver.ws;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.eventbus.Subscribe;
+import io.rudin.minetest.tileserver.service.EventBus;
+import org.eclipse.jetty.websocket.api.Session;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+@Singleton
+public class WebSocketUpdater {
+
+    @Inject
+    public WebSocketUpdater(EventBus eventBus){
+        this.eventBus = eventBus;
+        this.mapper = new ObjectMapper();
+    }
+
+    private final EventBus eventBus;
+
+    private final ObjectMapper mapper;
+
+    public void init(){
+        eventBus.register(this);
+    }
+
+    public static class EventContainer {
+        public String type;
+        public Object data;
+    }
+
+    @Subscribe void onPlayerMove(EventBus.PlayerMovedEvent e){
+        try {
+            EventContainer container = new EventContainer();
+            container.data = e;
+            container.type = "player-move";
+            String json = mapper.writeValueAsString(container);
+
+            for (Session session : WebSocketHandler.sessions) {
+                try {
+                    session.getRemote().sendString(json);
+
+                } catch (Exception e3) {
+                    //TODO
+                }
+            }
+        } catch (Exception e2){
+            e2.printStackTrace();
+        }
+
+    }
+
+    @Subscribe void onTileUpdate(EventBus.TileChangedEvent e){
+        try {
+            EventContainer container = new EventContainer();
+            container.data = e;
+            container.type = "tile-update";
+            String json = mapper.writeValueAsString(container);
+
+            for (Session session : WebSocketHandler.sessions) {
+                try {
+                    session.getRemote().sendString(json);
+
+                } catch (Exception e3) {
+                    //TODO
+                }
+            }
+        } catch (Exception e2){
+            e2.printStackTrace();
+        }
+
+    }
+
+}