Fix websockets, add test that starts, restarts and stops server

Author: JulianVennen

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

@@ -16,7 +16,8 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 
 public final class Server implements Initializable {
 
@@ -379,10 +380,7 @@ public CompletableFuture<Void> restart() throws IOException {
      * @return a future that completes when the server has reached one of the given statuses
      */
     public Future<Server> waitForStatus(Set<ServerStatus> statuses) {
-        if (this.webSocket == null) {
-            this.subscribe();
-        }
-        return webSocket.waitForStatus(statuses);
+        return this.subscribe().waitForStatus(statuses);
     }
 
     /**

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

@@ -6,24 +6,15 @@
 import com.exaroton.api.ws.subscriber.ServerStatusSubscriber;
 import org.jetbrains.annotations.ApiStatus;
 import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.*;
 
 @ApiStatus.Internal
 public final class WaitForStatusSubscriber implements ServerStatusSubscriber, Future<Server> {
     private final Set<ServerStatus> statuses;
-    private final ServerStatusStream stream;
-    @Nullable
-    private Server result;
-    private boolean cancelled;
-    @Nullable
-    private Throwable throwable;
-    @Nullable
-    private CountDownLatch latch;
+    private final CompletableFuture<Server> future;
 
     /**
      * Create a new future that will complete when the server has the specified status. This class will register itself
@@ -33,7 +24,9 @@ public final class WaitForStatusSubscriber implements ServerStatusSubscriber, Fu
      */
     public WaitForStatusSubscriber(Set<ServerStatus> statuses, ServerStatusStream stream) {
         this.statuses = Objects.requireNonNull(statuses);
-        this.stream = Objects.requireNonNull(stream);
+        ServerStatusStream stream1 = Objects.requireNonNull(stream);
+        this.future = new CompletableFuture<Server>()
+                .whenComplete((x, y) -> stream.removeSubscriber(this));
         stream.addSubscriber(this);
     }
 
@@ -42,94 +35,40 @@ public void handleStatusUpdate(Server oldServer, Server newServer) {
         try {
             if (newServer.hasStatus(statuses)) {
                 synchronized (this) {
-                    this.result = newServer;
+                    future.complete(newServer);
                 }
             }
         } catch (Throwable t) {
             synchronized (this) {
-                throwable = t;
+                future.completeExceptionally(t);
             }
         }
     }
 
     @Override
     public boolean cancel(boolean mayInterruptIfRunning) {
         synchronized (this) {
-            if (isDone()) {
-                return false;
-            }
-
-            cancelled = true;
+            return future.cancel(mayInterruptIfRunning);
         }
-        return true;
     }
 
     @Override
     public boolean isCancelled() {
-        return cancelled;
+        return future.isCancelled();
     }
 
     @Override
     public boolean isDone() {
-        return cancelled || throwable != null || result != null;
+        return future.isDone();
     }
 
     @Override
     public Server get() throws InterruptedException, ExecutionException {
-        var latch = getWaitingLatch();
-
-        if (latch.isPresent()) {
-            latch.get().await();
-        }
-
-        return getResult();
+        return future.get();
     }
 
     @Override
     public Server get(long timeout, @NotNull TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
-        var latch = getWaitingLatch();
-
-        if (latch.isPresent()) {
-            if (!latch.get().await(timeout, unit)) {
-                throw new TimeoutException();
-            }
-        }
-
-        stream.removeSubscriber(this);
-
-        return getResult();
-    }
-
-    /**
-     * Get the result or throw the corresponding exception. Future must have been completed before this is called.
-     * @return result
-     */
-    private Server getResult() throws ExecutionException {
-        if (cancelled) {
-            throw new CancellationException();
-        }
-
-        if (throwable != null) {
-            throw new ExecutionException(throwable);
-        }
-
-        return result;
-    }
-
-    /**
-     * Get the latch used to wait until the future is done
-     * @return latch or empty if the future is already done
-     */
-    private Optional<CountDownLatch> getWaitingLatch() {
-        synchronized (this) {
-            if (isDone()) {
-                return Optional.empty();
-            }
-
-            if (latch == null) {
-                latch = new CountDownLatch(1);
-            }
-            return Optional.of(latch);
-        }
+        return future.get(timeout, unit);
     }
 }

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

@@ -2,8 +2,11 @@
 
 import com.exaroton.api.server.Server;
 import com.exaroton.api.server.ServerStatus;
-import com.exaroton.api.ws.stream.*;
-import com.exaroton.api.ws.subscriber.*;
+import com.exaroton.api.ws.stream.ConsoleStream;
+import com.exaroton.api.ws.stream.ServerStatusStream;
+import com.exaroton.api.ws.stream.Stream;
+import com.exaroton.api.ws.stream.StreamType;
+import com.exaroton.api.ws.subscriber.ServerStatusSubscriber;
 import com.google.gson.Gson;
 import com.google.gson.JsonParser;
 import org.jetbrains.annotations.ApiStatus;
@@ -17,7 +20,9 @@
 import java.net.http.HttpClient;
 import java.net.http.WebSocket;
 import java.util.*;
-import java.util.concurrent.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Future;
 
 public final class WebSocketConnection implements WebSocket.Listener {
     /**
@@ -90,23 +95,6 @@ public WebSocketConnection(
         connect();
     }
 
-    /**
-     * subscribe to a stream if it is not already active
-     *
-     * @param name stream name
-     */
-    public void subscribe(@NotNull StreamType name) {
-        Objects.requireNonNull(name);
-
-        if (streams.containsKey(name.getStreamClass())) {
-            return;
-        }
-
-        Stream<?> stream = name.construct(this, gson);
-        this.streams.put(name.getStreamClass(), stream);
-        stream.start();
-    }
-
     /**
      * unsubscribe from a stream
      *
@@ -252,13 +240,13 @@ public <T> void removeStreamSubscriber(Class<? extends Stream<T>> c, T subscribe
     @ApiStatus.Internal
     public void unsubscribeFromEmptyStreams() {
         for (Stream<?> stream : streams.values()) {
-            if (!stream.hasSubscribers()) {
+            if (stream.hasNoSubscribers()) {
                 this.unsubscribe(stream.getType());
             }
         }
 
         // server status stream is always active
-        if (streams.size() == 1 && !getStream(ServerStatusStream.class).hasSubscribers()) {
+        if (streams.size() == 1 && getStream(ServerStatusStream.class).hasNoSubscribers()) {
             this.server.unsubscribe();
         }
     }
@@ -294,6 +282,7 @@ public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean
                     webSocket.sendText(x, true);
                 }
                 this.messages.clear();
+                break;
 
             default:
                 final StreamType name = StreamType.get(message.get("stream").getAsString());
@@ -336,6 +325,12 @@ public void run() {
         return null;
     }
 
+    @Override
+    @ApiStatus.Internal
+    public void onError(WebSocket webSocket, Throwable error) {
+        logger.error("Websocket connection to {} failed", uri, error);
+    }
+
     /**
      * send data once connection is ready
      *

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

@@ -1,7 +1,6 @@
 package com.exaroton.api.ws.stream;
 
 import com.exaroton.api.server.ServerStatus;
-import com.exaroton.api.ws.WaitForStatusSubscriber;
 import com.exaroton.api.ws.WebSocketConnection;
 import com.exaroton.api.ws.data.StreamData;
 import com.google.gson.Gson;
@@ -73,8 +72,8 @@ public void removeSubscriber(T subscriber) {
      * Check if this stream has subscribers
      * @return has subscribers?
      */
-    public boolean hasSubscribers() {
-        return !this.subscribers.isEmpty();
+    public boolean hasNoSubscribers() {
+        return this.subscribers.isEmpty();
     }
 
     /**

src/test/java/APIClientTest.java

@@ -1,17 +1,46 @@
 import com.exaroton.api.ExarotonClient;
 import com.exaroton.api.server.Server;
+import com.exaroton.api.server.ServerStatus;
 import org.jetbrains.annotations.NotNull;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 public abstract class APIClientTest {
     protected static final @NotNull String TEST_SERVER_ID = System.getenv("EXAROTON_TEST_SERVER");
-    protected final @NotNull ExarotonClient client = new ExarotonClient(System.getenv("EXAROTON_API_TOKEN"));
-    protected final @NotNull Server server = client.getServer(TEST_SERVER_ID);
+    protected ExarotonClient client;
+    protected Server server;
+
+    @BeforeEach
+    void setUp() {
+        client = new ExarotonClient(System.getenv("EXAROTON_API_TOKEN"));
+        server = client.getServer(TEST_SERVER_ID);
 
-    protected APIClientTest() {
         try {
             Thread.sleep(100);
         } catch (InterruptedException e) {
             throw new RuntimeException(e);
         }
     }
+
+    @AfterEach
+    void tearDown() throws IOException, ExecutionException, InterruptedException, TimeoutException {
+        server.fetch().join();
+
+        if (server.hasStatus(ServerStatus.RESTARTING, ServerStatus.LOADING, ServerStatus.PREPARING)) {
+            server.waitForStatus(ServerStatus.STARTING, ServerStatus.ONLINE, ServerStatus.OFFLINE, ServerStatus.CRASHED)
+                    .get(1, TimeUnit.MINUTES);
+        }
+
+        if (server.hasStatus(ServerStatus.ONLINE, ServerStatus.STARTING)) {
+            server.stop().join();
+            server.waitForStatus(ServerStatus.OFFLINE, ServerStatus.CRASHED).get(1, TimeUnit.MINUTES);
+        }
+
+        server.unsubscribe();
+    }
 }

src/test/java/ServerTest.java

@@ -1,10 +1,12 @@
 import com.exaroton.api.server.*;
-import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -133,8 +135,26 @@ void testGetPlayerLists() throws IOException {
     }
 
     @Test
-    void testStartServer() throws IOException {
-        assertEquals(ServerStatus.OFFLINE, server.getStatus());
+    void testStartRestartStopServer() throws IOException, ExecutionException, InterruptedException, TimeoutException {
+        server.fetch().join();
+        assertTrue(server.hasStatus(ServerStatus.OFFLINE, ServerStatus.CRASHED));
+
+        // Start server
         server.start().join();
+        server.waitForStatus(ServerStatus.ONLINE).get(3, TimeUnit.MINUTES);
+        assertEquals(ServerStatus.ONLINE, server.getStatus());
+
+        // Restart Server
+        server.restart().join();
+        server.waitForStatus(ServerStatus.RESTARTING).get(1, TimeUnit.MINUTES);
+        assertEquals(ServerStatus.RESTARTING, server.getStatus());
+
+        server.waitForStatus(ServerStatus.ONLINE).get(3, TimeUnit.MINUTES);
+        assertEquals(ServerStatus.ONLINE, server.getStatus());
+
+        // Stop Server
+        server.stop().join();
+        server.waitForStatus(ServerStatus.OFFLINE, ServerStatus.CRASHED).get(3, TimeUnit.MINUTES);
+        assertTrue(server.hasStatus(ServerStatus.OFFLINE, ServerStatus.CRASHED), "Expected server to be offline or crashed");
     }
 }