More WS refactors

Author: JulianVennen

CHANGELOG.md

@@ -40,8 +40,12 @@ string or object.
 The `java-websocket` library has been replaced by the built-in Java 11 websocket implementation. Unless you were using
 the `WebSocketClient` or `WebSocketManager` classes directly this shouldn't require any changes to your code.
 
+`Server#subscribe` and `Server#unsubscribe` now accept the `StreamName` enum instead of any string
+
+The debug/error handler methods have been removed. Errors and debug messages are now  exclusively logged using SLF4J
+
+
 ### Other
-- `Server#subscribe` and `Server#unsubscribe` now accept the `StreamName` enum instead of any string
 - Arrays have been Replaced by Collection's in almost all places
 - Config options now return a generic type instead of `Object`
 - Renamed `ServerFile#getInfo` to `ServerFile#get`
@@ -50,8 +54,6 @@ the `WebSocketClient` or `WebSocketManager` classes directly this shouldn't requ
 - Removed `ExarotonClient#getBaseUrl()` and `ExarotonClient#createConnection(String, String)`
 - Many classes are now final
 - Added `ApiStatus` annotations to many classes and methods
-- Removed debug/error handler from `WebsocketManager` and `WebsocketClient`. Errors and debug messages are now
-  exclusively logged using SLF4J
 - Removed `ExarotonClient#getApiToken()`
 
 ## Improvements

src/main/java/com/exaroton/api/server/Server.java

@@ -4,7 +4,7 @@
 import com.exaroton.api.Initializable;
 import com.exaroton.api.request.server.*;
 import com.exaroton.api.ws.WebSocketConnection;
-import com.exaroton.api.ws.stream.StreamName;
+import com.exaroton.api.ws.stream.StreamType;
 import com.exaroton.api.ws.subscriber.*;
 import com.google.gson.Gson;
 import org.jetbrains.annotations.ApiStatus;
@@ -461,8 +461,8 @@ public void subscribe() {
      *
      * @param streams stream names
      */
-    public void subscribe(StreamName... streams) {
-        for (StreamName stream : streams) {
+    public void subscribe(StreamType... streams) {
+        for (StreamType stream : streams) {
             if (this.webSocket == null) {
                 this.subscribe();
             }
@@ -485,9 +485,9 @@ public void unsubscribe() {
      *
      * @param streams stream names
      */
-    public void unsubscribe(StreamName... streams) {
+    public void unsubscribe(StreamType... streams) {
         if (this.webSocket == null) throw new RuntimeException("No websocket connection active.");
-        for (StreamName stream : streams) {
+        for (StreamType stream : streams) {
             this.webSocket.unsubscribe(stream);
         }
     }

src/main/java/com/exaroton/api/ws/WebSocketConnection.java

@@ -2,11 +2,9 @@
 
 import com.exaroton.api.server.Server;
 import com.exaroton.api.server.ServerStatus;
-import com.exaroton.api.ws.data.*;
 import com.exaroton.api.ws.stream.*;
 import com.exaroton.api.ws.subscriber.*;
 import com.google.gson.Gson;
-import com.google.gson.JsonObject;
 import com.google.gson.JsonParser;
 import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
@@ -57,7 +55,7 @@ public final class WebSocketConnection implements WebSocket.Listener, Closeable
     /**
      * active streams
      */
-    private final Map<StreamName, Stream> streams = new HashMap<>();
+    private final Map<Class<? extends Stream<?>>, Stream<?>> streams = new HashMap<>();
 
     @NotNull
     private final String apiToken;
@@ -88,7 +86,7 @@ public WebSocketConnection(
         this.uri = uri;
         this.apiToken = apiToken;
         this.server = Objects.requireNonNull(server);
-        this.streams.put(StreamName.STATUS, new Stream(this, this.gson, StreamName.STATUS));
+        this.streams.put(ServerStatusStream.class, new ServerStatusStream(this, this.gson).setServer(server));
 
         connect();
     }
@@ -98,21 +96,15 @@ public WebSocketConnection(
      *
      * @param name stream name
      */
-    public void subscribe(@NotNull StreamName name) {
+    public void subscribe(@NotNull StreamType name) {
         Objects.requireNonNull(name);
 
-        if (streams.containsKey(name)) {
+        if (streams.containsKey(name.getStreamClass())) {
             return;
         }
 
-        Stream stream;
-        if (name == StreamName.CONSOLE) {
-            stream = new ConsoleStream(this, this.gson);
-        } else {
-            stream = new Stream(this, this.gson, name);
-        }
-
-        this.streams.put(name, stream);
+        Stream<?> stream = name.construct(this, gson);
+        this.streams.put(name.getStreamClass(), stream);
         stream.start();
     }
 
@@ -121,13 +113,13 @@ public void subscribe(@NotNull StreamName name) {
      *
      * @param name stream name
      */
-    public void unsubscribe(@NotNull StreamName name) {
+    public void unsubscribe(@NotNull StreamType name) {
         Objects.requireNonNull(name);
 
-        Stream stream = this.streams.get(name);
+        Stream<?> stream = this.streams.get(name.getStreamClass());
         if (stream != null) {
             stream.stop();
-            this.streams.remove(name);
+            this.streams.remove(name.getStreamClass());
         }
     }
 
@@ -137,7 +129,7 @@ public void unsubscribe(@NotNull StreamName name) {
      * @param subscriber instance of class handling server status changes
      */
     public void addServerStatusSubscriber(ServerStatusSubscriber subscriber) {
-        this.streams.get(StreamName.STATUS).subscribers.add(subscriber);
+        this.addStreamSubscriber(ServerStatusStream.class, subscriber);
     }
 
     /**
@@ -146,7 +138,7 @@ public void addServerStatusSubscriber(ServerStatusSubscriber subscriber) {
      * @param subscriber instance of class handling new console lines
      */
     public void addConsoleSubscriber(ConsoleSubscriber subscriber) {
-        this.addStreamSubscriber(StreamName.CONSOLE, subscriber);
+        this.addStreamSubscriber(ConsoleStream.class, subscriber);
     }
 
     /**
@@ -155,7 +147,7 @@ public void addConsoleSubscriber(ConsoleSubscriber subscriber) {
      * @param subscriber instance of class handling heap data
      */
     public void addHeapSubscriber(HeapSubscriber subscriber) {
-        this.addStreamSubscriber(StreamName.HEAP, subscriber);
+        this.addStreamSubscriber(HeapStream.class, subscriber);
     }
 
     /**
@@ -164,7 +156,7 @@ public void addHeapSubscriber(HeapSubscriber subscriber) {
      * @param subscriber instance of class handling stats
      */
     public void addStatsSubscriber(StatsSubscriber subscriber) {
-        this.addStreamSubscriber(StreamName.STATS, subscriber);
+        this.addStreamSubscriber(StatsStream.class, subscriber);
     }
 
     /**
@@ -173,7 +165,7 @@ public void addStatsSubscriber(StatsSubscriber subscriber) {
      * @param subscriber instance of class handling stats
      */
     public void addTickSubscriber(TickSubscriber subscriber) {
-        this.addStreamSubscriber(StreamName.TICK, subscriber);
+        this.addStreamSubscriber(TickStream.class, subscriber);
     }
 
     /**
@@ -183,9 +175,9 @@ public void addTickSubscriber(TickSubscriber subscriber) {
      * @return was the command executed
      */
     public boolean executeCommand(String command) {
-        Stream s = this.streams.get(StreamName.CONSOLE);
-        if (s instanceof ConsoleStream) {
-            ((ConsoleStream) s).executeCommand(command);
+        ConsoleStream stream = this.getStream(ConsoleStream.class);
+        if (stream != null) {
+            stream.executeCommand(command);
             return true;
         }
 
@@ -236,9 +228,17 @@ public CompletableFuture<Boolean> serverHasStatus(Set<ServerStatus> status) {
         return future.thenApply(s -> s.hasStatus(status));
     }
 
-    private void addStreamSubscriber(StreamName name, Subscriber subscriber) {
-        if (!this.streams.containsKey(name)) throw new RuntimeException("There is no active stream for: " + name);
-        this.streams.get(name).subscribers.add(subscriber);
+    private <T extends Subscriber> void addStreamSubscriber(Class<? extends Stream<T>> c, T subscriber) {
+        if (!this.streams.containsKey(c)) {
+            throw new IllegalStateException("There is no active stream for: " + c);
+        }
+
+        getStream(c).addSubscriber(subscriber);
+    }
+
+    private <T extends Stream<?>> T getStream(Class<T> c) {
+        @SuppressWarnings("unchecked") T stream = (T) this.streams.get(c);
+        return stream;
     }
 
     private void connect() {
@@ -284,78 +284,21 @@ public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean
                 this.messages.clear();
 
             default:
-                handleData(type, message);
+                final StreamType name = StreamType.get(message.get("stream").getAsString());
+                final Stream<?> stream = streams.get(name.getStreamClass());
+                if (stream != null) {
+                    stream.onMessage(type, message);
+                }
         }
 
         return null;
     }
 
-    /**
-     * handle websocket data
-     *
-     * @param type    message type
-     * @param message raw message
-     */
-    private void handleData(String type, JsonObject message) {
-        final StreamName name = StreamName.get(type);
-        final Stream stream = streams.get(name);
-
-        if (stream == null) {
-            return;
-        }
-
-        switch (type) {
-            case "status":
-                Server oldServer = new Server(server.getClient(), gson, server.getId()).setFromObject(server);
-                this.server.setFromObject(gson.fromJson(message, ServerStatusStreamData.class).getData());
-
-                //start/stop streams based on status
-                for (Stream s : streams.values()) {
-                    s.onStatusChange();
-                }
-
-                for (Subscriber subscriber : stream.subscribers) {
-                    if (subscriber instanceof ServerStatusSubscriber) {
-                        ((ServerStatusSubscriber) subscriber).statusUpdate(oldServer, this.server);
-                    }
-                }
-                break;
-
-            case "line":
-                String line = gson.fromJson(message, ConsoleStreamData.class).getData();
-                for (Subscriber subscriber : stream.subscribers) {
-                    if (subscriber instanceof ConsoleSubscriber) {
-                        ((ConsoleSubscriber) subscriber).line(line);
-                    }
-                }
-                break;
-
-            case "heap":
-                HeapUsage usage = gson.fromJson(message, HeapStreamData.class).getData();
-                for (Subscriber subscriber : stream.subscribers) {
-                    if (subscriber instanceof HeapSubscriber) {
-                        ((HeapSubscriber) subscriber).heap(usage);
-                    }
-                }
-                break;
-
-            case "stats":
-                StatsData stats = gson.fromJson(message, StatsStreamData.class).getData();
-                for (Subscriber subscriber : stream.subscribers) {
-                    if (subscriber instanceof StatsSubscriber) {
-                        ((StatsSubscriber) subscriber).stats(stats);
-                    }
-                }
-                break;
-
-            case "tick":
-                TickData tick = gson.fromJson(message, TickStreamData.class).getData();
-                for (Subscriber subscriber : stream.subscribers) {
-                    if (subscriber instanceof TickSubscriber) {
-                        ((TickSubscriber) subscriber).tick(tick);
-                    }
-                }
-                break;
+    @ApiStatus.Internal
+    public void onStatusChange() {
+        // start/stop streams based on status
+        for (Stream<?> s : streams.values()) {
+            s.onStatusChange();
         }
     }
 
@@ -364,6 +307,10 @@ private void handleData(String type, JsonObject message) {
     public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason) {
         logger.info("Websocket connection to {} closed: {} {}", uri, statusCode, reason);
 
+        for (Stream<?> stream : streams.values()) {
+            stream.onDisconnected();
+        }
+
         if (this.shouldAutoReconnect()) {
             reconnectTimer = new Timer();
             logger.debug("Reconnecting in 5s");
@@ -405,4 +352,8 @@ public void close() {
             this.client.sendClose(0, "unsubscribe");
         }
     }
+
+    public boolean isReady() {
+        return ready;
+    }
 }

src/main/java/com/exaroton/api/ws/stream/ConsoleStream.java

@@ -1,13 +1,17 @@
 package com.exaroton.api.ws.stream;
 
 import com.exaroton.api.ws.WebSocketConnection;
+import com.exaroton.api.ws.data.ConsoleStreamData;
+import com.exaroton.api.ws.subscriber.ConsoleSubscriber;
 import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
 
-public class ConsoleStream extends Stream {
-
+public final class ConsoleStream extends Stream<ConsoleSubscriber> {
+    @ApiStatus.Internal
     public ConsoleStream(@NotNull WebSocketConnection ws, @NotNull Gson gson) {
-        super(ws, gson, StreamName.CONSOLE);
+        super(ws, gson);
     }
 
     /**
@@ -17,4 +21,20 @@ public ConsoleStream(@NotNull WebSocketConnection ws, @NotNull Gson gson) {
     public void executeCommand(String command) {
         this.send("command", command);
     }
+
+    @Override
+    protected void onDataMessage(String type, JsonObject message) {
+        switch (type) {
+            case "line":
+                String line = gson.fromJson(message, ConsoleStreamData.class).getData();
+                for (ConsoleSubscriber subscriber : subscribers) {
+                    subscriber.line(line);
+                }
+        }
+    }
+
+    @Override
+    protected StreamType getType() {
+        return StreamType.CONSOLE;
+    }
 }

src/main/java/com/exaroton/api/ws/stream/HeapStream.java

@@ -0,0 +1,35 @@
+package com.exaroton.api.ws.stream;
+
+import com.exaroton.api.ws.WebSocketConnection;
+import com.exaroton.api.ws.data.HeapStreamData;
+import com.exaroton.api.ws.data.HeapUsage;
+import com.exaroton.api.ws.subscriber.HeapSubscriber;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.jetbrains.annotations.ApiStatus;
+import org.jetbrains.annotations.NotNull;
+
+public final class HeapStream extends Stream<HeapSubscriber> {
+    @ApiStatus.Internal
+    public HeapStream(@NotNull WebSocketConnection ws, @NotNull Gson gson) {
+        super(ws, gson);
+    }
+
+    @Override
+    protected void onDataMessage(String type, JsonObject message) {
+        switch (type) {
+            case "heap":
+                HeapUsage usage = gson.fromJson(message, HeapStreamData.class).getData();
+
+                for (HeapSubscriber subscriber : subscribers) {
+                    subscriber.heap(usage);
+                }
+                break;
+        }
+    }
+
+    @Override
+    protected StreamType getType() {
+        return StreamType.HEAP;
+    }
+}