[SCB-2894] optimize WebSocket closing race condition and add WebSocket status

Author: yhs0092

swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/ws/AbstractBaseWebSocket.java

@@ -19,12 +19,22 @@
 
 import java.util.concurrent.CompletableFuture;
 
+import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
+
 /**
  * AbstractBaseWebSocket
  */
 public abstract class AbstractBaseWebSocket implements WebSocket {
   private WebSocketAdapter webSocketAdapter;
 
+  private Status status = Status.CREATED;
+
+  private CompletableFuture<Void> closeFuture;
+
+  private Short closeStatusCode;
+
+  private String closeReason;
+
   @Override
   public CompletableFuture<Void> sendMessage(WebSocketMessage<?> message) {
     return webSocketAdapter.sendMessage(message);
@@ -37,21 +47,39 @@ public CompletableFuture<Void> sendFrame(WebSocketFrame frame) {
 
   @Override
   public CompletableFuture<Void> close() {
-    return webSocketAdapter.close((short) 1000, "NORMAL");
+    return this.close((short) WebSocketCloseStatus.NORMAL_CLOSURE.code(),
+        WebSocketCloseStatus.NORMAL_CLOSURE.reasonText());
   }
 
   @Override
   public CompletableFuture<Void> close(Short closeStatusCode, String closeReason) {
-    return webSocketAdapter.close(closeStatusCode, closeReason);
+    synchronized (this) {
+      if (status == Status.WAITING_TO_CLOSE || status == Status.CLOSING || status == Status.CLOSED) {
+        return CompletableFuture.completedFuture(null);
+      }
+      status = Status.WAITING_TO_CLOSE;
+      this.closeStatusCode = closeStatusCode;
+      this.closeReason = closeReason;
+      if (webSocketAdapter == null) {
+        // the case that close when WebSocket still not complete handshake
+        closeFuture = new CompletableFuture<>();
+        return closeFuture;
+      }
+    }
+    status = Status.CLOSING;
+    return webSocketAdapter.close(closeStatusCode, closeReason)
+        .whenComplete((v, t) -> status = Status.CLOSED);
   }
 
   @Override
   public void pause() {
+    status = Status.PAUSED;
     webSocketAdapter.pause();
   }
 
   @Override
   public void resume() {
+    status = Status.RUNNING;
     webSocketAdapter.resume();
   }
 
@@ -78,4 +106,29 @@ public void onWriteQueueDrain() {
   public void setWebSocketAdapter(WebSocketAdapter webSocketAdapter) {
     this.webSocketAdapter = webSocketAdapter;
   }
+
+  public void startWorking() {
+    synchronized (this) {
+      if (status == Status.WAITING_TO_CLOSE) {
+        status = Status.CLOSING;
+        webSocketAdapter.close(closeStatusCode, closeReason)
+            .whenComplete((v, t) -> {
+              status = Status.CLOSED;
+              if (t != null) {
+                closeFuture.completeExceptionally(t);
+              } else {
+                closeFuture.complete(null);
+              }
+            });
+        return;
+      }
+      status = Status.RUNNING;
+    }
+    onConnectionReady();
+  }
+
+  @Override
+  public Status getStatus() {
+    return status;
+  }
 }

swagger/swagger-invocation/invocation-core/src/main/java/org/apache/servicecomb/swagger/invocation/ws/WebSocket.java

@@ -56,4 +56,15 @@ default void onFrame(WebSocketFrame frame) {
   boolean writeQueueFull();
 
   void onWriteQueueDrain();
+
+  Status getStatus();
+
+  enum Status {
+    CREATED,
+    RUNNING,
+    PAUSED,
+    WAITING_TO_CLOSE,
+    CLOSING,
+    CLOSED
+  }
 }

transports/transport-rest/transport-rest-vertx/src/main/java/org/apache/servicecomb/transport/rest/vertx/ws/VertxWebSocketAdaptor.java

@@ -162,7 +162,7 @@ private void linkVertxDrainHandler() {
 
   private void startWorking() {
     scheduleTask(
-        bizWebSocket::onConnectionReady);
+        bizWebSocket::startWorking);
   }
 
   private void scheduleTask(Runnable task) {