Repository: GoogleCloudPlatform/pgadapter
Pull request #59 (2022-03-09 18:55:41)

chore: model ProxyServer as an ApiService
Use a standard AbstractApiService to model the ProxyServer
instead of keeping track of the state of the server manually.
This reduces code complexity by relying on a standard service
that is included in the standard Google libraries.


Patch:
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
index cd01d9aa..a624476f 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import com.google.api.core.AbstractApiService;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.SpannerException;
@@ -22,7 +23,6 @@ import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
-import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
@@ -33,23 +33,20 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.logging.Level;
 import java.util.logging.Logger;
-import javax.annotation.concurrent.GuardedBy;
 
 /**
  * The proxy server listens for incoming client connections and starts a new {@link
  * ConnectionHandler} for each incoming connection.
  */
-public class ProxyServer extends Thread {
+public class ProxyServer extends AbstractApiService {
 
   private static final Logger logger = Logger.getLogger(ProxyServer.class.getName());
   private final OptionsMetadata options;
   private final Properties properties;
   private final List<ConnectionHandler> handlers = new LinkedList<>();
 
-  @GuardedBy("itself")
-  private volatile ServerStatus status = ServerStatus.NEW;
-
   private ServerSocket serverSocket;
+  private int localPort;
 
   /**
    * Instantiates the ProxyServer from CLI-gathered metadata.
@@ -57,8 +54,8 @@ public class ProxyServer extends Thread {
    * @param optionsMetadata Resulting metadata from CLI.
    */
   public ProxyServer(OptionsMetadata optionsMetadata) {
-    super("spanner-postgres-adapter-proxy-port-" + optionsMetadata.getProxyPort());
     this.options = optionsMetadata;
+    this.localPort = optionsMetadata.getProxyPort();
     this.properties = new Properties();
     addConnectionProperties();
   }
@@ -72,8 +69,8 @@ public class ProxyServer extends Thread {
    *     supported properties).
    */
   public ProxyServer(OptionsMetadata optionsMetadata, Properties properties) {
-    super("spanner-postgres-adapter-proxy-port-" + optionsMetadata.getProxyPort());
     this.options = optionsMetadata;
+    this.localPort = optionsMetadata.getProxyPort();
     this.properties = properties;
     addConnectionProperties();
   }
@@ -86,48 +83,50 @@ public class ProxyServer extends Thread {
 
   /** Starts the server by running the thread runnable and setting status. */
   public void startServer() {
-    synchronized (this.status) {
-      Preconditions.checkState(this.status == ServerStatus.NEW);
+    startAsync();
+    awaitRunning();
+    logger.log(Level.INFO, "Server started on port {0}", String.valueOf(getLocalPort()));
+  }
+
+  @Override
+  protected void doStart() {
+    Thread listenerThread =
+        new Thread("spanner-postgres-adapter-proxy-listener") {
+          @Override
+          public void run() {
+            try {
+              runServer();
+            } catch (IOException e) {
+              logger.log(
+                  Level.WARNING,
+                  "Server on port {0} stopped by exception: {1}",
+                  new Object[] {getLocalPort(), e});
+            }
+          }
+        };
+    listenerThread.start();
+  }
+
+  @Override
+  protected void doStop() {
+    try {
+      logger.log(Level.INFO, "Server on port {0} is stopping", String.valueOf(getLocalPort()));
+      this.serverSocket.close();
+      logger.log(Level.INFO, "Server socket on port {0} closed", String.valueOf(getLocalPort()));
+    } catch (IOException exception) {
       logger.log(
-          Level.INFO,
-          "Server is starting on port {0}",
-          String.valueOf(this.options.getProxyPort()));
-      this.status = ServerStatus.STARTING;
-      start();
-      // TODO: Consider changing the server to use ApiService, so we don't have to manage the state
-      //  of the server manually.
-      while (this.status == ServerStatus.STARTING) {
-        Thread.yield();
-      }
-      logger.log(Level.INFO, "Server started on port {0}", String.valueOf(getLocalPort()));
+          Level.WARNING,
+          "Closing server socket on port {0} failed: {1}",
+          new Object[] {String.valueOf(getLocalPort()), exception});
     }
   }
 
   /** Safely stops the server (iff started), closing specific socket and cleaning up. */
   public void stopServer() {
-    synchronized (status) {
-      Preconditions.checkState(status == ServerStatus.STARTED, "Server is not in state Started");
-      try {
-        logger.log(
-            Level.INFO,
-            "Server on port {0} is stopping",
-            String.valueOf(this.options.getProxyPort()));
-        this.status = ServerStatus.STOPPING;
-        this.serverSocket.close();
-        logger.log(
-            Level.INFO,
-            "Server socket on port {0} closed",
-            String.valueOf(this.options.getProxyPort()));
-      } catch (IOException e) {
-        logger.log(
-            Level.WARNING,
-            "Closing server socket on port {0} failed: {1}",
-            new Object[] {String.valueOf(this.options.getProxyPort()), e});
-      }
-    }
+    stopAsync();
+    awaitTerminated();
   }
 
-  @Override
   public void run() {
     try {
       runServer();
@@ -135,7 +134,7 @@ public class ProxyServer extends Thread {
       logger.log(
           Level.WARNING,
           "Server on port {0} stopped by exception: {1}",
-          new Object[] {this.options.getProxyPort(), e});
+          new Object[] {getLocalPort(), e});
     }
   }
 
@@ -146,25 +145,31 @@ public class ProxyServer extends Thread {
    */
   void runServer() throws IOException {
     this.serverSocket = new ServerSocket(this.options.getProxyPort());
-    this.status = ServerStatus.STARTED;
+    this.localPort = serverSocket.getLocalPort();
+    notifyStarted();
     try {
-      while (this.status == ServerStatus.STARTED) {
+      while (isRunning()) {
         createConnectionHandler(this.serverSocket.accept());
       }
     } catch (SocketException e) {
-      // This is a normal exception, as this will occur when Server#stopServer() is called.
-      logger.log(
-          Level.INFO,
-          "Socket exception on port {0}: {1}. This is normal when the server is stopped.",
-          new Object[] {getLocalPort(), e});
+      if (state() == State.STOPPING) {
+        // This is a normal exception, as this will occur when Server#stopServer() is called.
+        logger.log(
+            Level.FINE,
+            "Socket exception on port {0}: {1} while stopping the server. This is normal.",
+            new Object[] {getLocalPort(), e});
+      } else {
+        logger.log(
+            Level.SEVERE, "Socket exception on port {0}: {1}.", new Object[] {getLocalPort(), e});
+      }
     } catch (SpannerException e) {
       logger.log(
           Level.SEVERE,
           "Something went wrong in establishing a Spanner connection: {0}",
           e.getMessage());
     } finally {
-      this.status = ServerStatus.STOPPED;
       logger.log(Level.INFO, "Socket on port {0} stopped", getLocalPort());
+      notifyStopped();
     }
   }
 
@@ -222,21 +227,12 @@ public class ProxyServer extends Thread {
   }
 
   public int getLocalPort() {
-    if (this.serverSocket != null) {
-      return this.serverSocket.getLocalPort();
-    }
-    return -1;
-  }
-
-  public ServerStatus getServerStatus() {
-    synchronized (this.status) {
-      return this.status;
-    }
+    return localPort;
   }
 
   @Override
   public String toString() {
-    return "ProxyServer[" + getName() + "]";
+    return String.format("ProxyServer[port: %d]", getLocalPort());
   }
 
   /**
@@ -313,13 +309,4 @@ public class ProxyServer extends Thread {
       return code;
     }
   }
-
-  /** Possible states of a {@link Server}. */
-  public enum ServerStatus {
-    NEW,
-    STARTING,
-    STARTED,
-    STOPPING,
-    STOPPED
-  }
 }


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
index f1ff13aa..16b99b95 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
@@ -299,11 +299,6 @@ abstract class AbstractMockServerTest {
   @AfterClass
   public static void stopMockSpannerAndPgAdapterServers() throws Exception {
     pgServer.stopServer();
-    while (pgServer.isAlive()) {
-      // TODO: Remove once the pgServer.stopServer() is blocking until it has actually stopped (or
-      // there is some other way to block until it has stopped).
-      Thread.sleep(1L);
-    }
     try {
       SpannerPool.closeSpannerPool();
     } catch (SpannerException e) {
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java b/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
index 1406b23d..1d00b3e7 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
@@ -14,7 +14,6 @@
 
 package com.google.cloud.spanner.pgadapter;
 
-import static com.google.cloud.spanner.pgadapter.ProxyServer.ServerStatus;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertThat;
@@ -45,6 +44,7 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
+import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -342,15 +342,7 @@ public final class PgAdapterTestEnv {
   }
 
   void waitForServer(ProxyServer server) throws Exception {
-    // Wait for up to 1 second if server has not started.
-    for (int i = 0; i < 10; ++i) {
-      if (server.getServerStatus() == ServerStatus.STARTED) {
-        return;
-      }
-      Thread.sleep(100);
-    }
-    // Throw exception if server has still not started.
-    throw new IllegalStateException("ProxyServer failed to start.");
+    server.awaitRunning(1L, TimeUnit.SECONDS);
   }
 
   void initializeConnection(DataOutputStream out) throws Exception {