Rework transport states to make sure server transports are associated with at most one connection. The connection can either be established at the formation of the server comms stack or associated with an already formed reconnection server stack during the reconnect window

Author: Myron Scott

common/src/main/java/com/tc/net/protocol/transport/ClientMessageTransport.java

@@ -490,7 +490,7 @@ private void handshakeConnection() throws TCTimeoutException, MaxConnectionsExce
     handleHandshakeError(result);
     initConnectionID(result.synAck.getConnectionId());
     sendAck();
-    log("Handshake is complete");
+    logger.debug("Handshake is complete");
   }
 
   private String getMaxConnectionsExceededMessage(int maxConnections) {

common/src/main/java/com/tc/net/protocol/transport/MessageTransportBase.java

@@ -190,13 +190,13 @@ public final void send(TCNetworkMessage message) throws IOException {
     }
   }
 
-  // Do not override this method. Not a final method, as a test class is deriving it
   @Override
   public void sendToConnection(TCNetworkMessage message) throws IOException {
     if (message == null) throw new AssertionError("Attempt to send a null message.");
     if (!status.isEnd()) {
       connection.putMessage(message);
     } else {
+      message.complete();
       throw new IOException("Couldn't send message status: " + status);
     }
   }
@@ -211,51 +211,31 @@ public boolean isConnected() {
   }
 
   @Override
-  public final void attachNewConnection(TCConnection newConnection) throws IllegalReconnectException {
-    getConnectionAttacher().attachNewConnection(this.connectionCloseEvent.getAndSet(null), this.connection,
+  public void attachNewConnection(TCConnection newConnection) throws IllegalReconnectException {
+    attachNewConnection(this.connectionCloseEvent.getAndSet(null), this.connection,
                                                 newConnection);
   }
 
-  protected ConnectionAttacher getConnectionAttacher() {
-    return new DefaultConnectionAttacher(this, getLogger());
-  }
-
-  protected interface ConnectionAttacher {
-    public void attachNewConnection(TCConnectionEvent closeEvent, TCConnection oldConnection, TCConnection newConnection);
-  }
-
-  private static final class DefaultConnectionAttacher implements ConnectionAttacher {
-
-    private final MessageTransportBase transport;
-    private final Logger logger;
-
-    private DefaultConnectionAttacher(MessageTransportBase transport, Logger logger) {
-      this.transport = transport;
-      this.logger = logger;
-    }
-
-    @Override
-    public void attachNewConnection(TCConnectionEvent closeEvent, TCConnection oldConnection, TCConnection newConnection) {
-      Assert.assertNotNull(oldConnection);
-      if (closeEvent == null || closeEvent.getSource() != oldConnection) {
-        // We either didn't receive a close event or we received a close event
-        // from a connection that isn't our current connection.
-        if (transport.isConnected()) {
-          // DEV-1689 : Don't bother for connections which actually didn't make up to Transport Establishment.
-          this.transport.status.reset();
-          this.transport.fireTransportDisconnectedEvent();
-          this.transport.getConnection().asynchClose();
-        } else {
-          logger.warn("Old connection " + oldConnection + "might not have been Transport Established ");
-        }
-      }
-      // remove the transport as a listener for the old connection
-      if (oldConnection != null && oldConnection != transport.getConnection()) {
-        oldConnection.removeListener(transport);
+  private void attachNewConnection(TCConnectionEvent closeEvent, TCConnection oldConnection, TCConnection newConnection) {
+    Assert.assertNotNull(oldConnection);
+    if (closeEvent == null || closeEvent.getSource() != oldConnection) {
+      // We either didn't receive a close event or we received a close event
+      // from a connection that isn't our current connection.
+      if (isConnected()) {
+        // DEV-1689 : Don't bother for connections which actually didn't make up to Transport Establishment.
+        status.reset();
+        fireTransportDisconnectedEvent();
+        getConnection().asynchClose();
+      } else {
+        logger.warn("Old connection " + oldConnection + "might not have been Transport Established ");
       }
-      // set the new connection to the current connection.
-      transport.wireNewConnection(newConnection);
     }
+    // remove the transport as a listener for the old connection
+    if (oldConnection != null && oldConnection != getConnection()) {
+      oldConnection.removeListener(this);
+    }
+    // set the new connection to the current connection.
+    wireNewConnection(newConnection);
   }
 
   /*********************************************************************************************************************
@@ -410,15 +390,6 @@ public TCByteBufferOutputStream createOutput() {
     }
   }
 
-  void log(String msg) {
-    if (!getProductID().isInternal()) {
-      getLogger().info(msg);
-    } else {
-      getLogger().debug(msg);
-    }
-
-  }
-
   @Override
   public Map<String, ?> getStateMap() {
     Map<String, Object> map = new LinkedHashMap<>();

common/src/main/java/com/tc/net/protocol/transport/MessageTransportState.java

@@ -28,12 +28,15 @@ public boolean isAlive() {
       return false;
     }
   },
-  
-  STATE_START_OPEN("START_OPEN"),
-  
+    
   STATE_CONNECTED("CONNECTED"),
 
-  STATE_RESTART("RESTART"),
+  STATE_RESTART("RESTART") {
+    @Override
+    public boolean isAlive() {
+      return false;
+    }
+  },
 
   /**
    * XXX: Move to client state machine SYN message sent, waiting for reply

common/src/main/java/com/tc/net/protocol/transport/MessageTransportStatus.java

@@ -23,23 +23,19 @@
 import com.tc.util.Assert;
 
 class MessageTransportStatus {
-  private final MessageTransportState initial;
   private MessageTransportState state;
   private final Logger logger;
-  private volatile boolean isEstablished = false;
 
   MessageTransportStatus(MessageTransportState initialState, Logger logger) {
-    this.initial = initialState;
     this.state = initialState;
     this.logger = logger;
   }
 
   void reset() {
-    stateChange(initial);
+    stateChange(MessageTransportState.STATE_START);
   }
 
   private synchronized void stateChange(MessageTransportState newState) {
-    isEstablished = false;
     if (logger.isDebugEnabled()) {
       logger.debug("Changing from " + state.toString() + " to " + newState.toString());
     }
@@ -78,19 +74,12 @@ void end() {
   }
 
   private synchronized boolean checkState(MessageTransportState check) {
-    if (check == MessageTransportState.STATE_ESTABLISHED && state == MessageTransportState.STATE_ESTABLISHED) {
-      isEstablished = true;
-    }
     return this.state.equals(check);
   }
 
   boolean isStart() {
     return checkState(MessageTransportState.STATE_START);
   }
-
-  boolean isStartOpen() {
-    return checkState(MessageTransportState.STATE_START_OPEN);
-  }
 
   boolean isRestart() {
     return checkState(MessageTransportState.STATE_RESTART);
@@ -101,11 +90,7 @@ boolean isSynSent() {
   }
 
   boolean isEstablished() {
-    if (isEstablished) {
-      return true;
-    } else {
-      return checkState(MessageTransportState.STATE_ESTABLISHED);
-    }
+    return checkState(MessageTransportState.STATE_ESTABLISHED);
   }
 
   boolean isDisconnected() {

common/src/main/java/com/tc/net/protocol/transport/ServerMessageTransport.java

@@ -22,7 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 import com.tc.net.core.TCConnection;
-import com.tc.net.core.event.TCConnectionEvent;
+import com.tc.net.protocol.IllegalReconnectException;
 import com.tc.net.protocol.NetworkStackID;
 import com.tc.util.Assert;
 
@@ -44,16 +44,20 @@ public ServerMessageTransport(TransportHandshakeErrorHandler handshakeErrorHandl
   public ServerMessageTransport(TCConnection conn,
                                 TransportHandshakeErrorHandler handshakeErrorHandler,
                                 TransportHandshakeMessageFactory messageFactory) {
-    super(MessageTransportState.STATE_START_OPEN, handshakeErrorHandler, messageFactory, smtLogger);
+    super(MessageTransportState.STATE_START, handshakeErrorHandler, messageFactory, smtLogger);
     Assert.assertNotNull(conn);
     wireNewConnection(conn);
   }
 
   @Override
-  protected ConnectionAttacher getConnectionAttacher() {
-    if (this.status.isRestart()) {
-      return new RestartConnectionAttacher();
-    } else return super.getConnectionAttacher();
+  public void attachNewConnection(TCConnection newConnection) throws IllegalReconnectException {
+    if (!this.status.isRestart()) {
+      // servers transports can only restart once
+      throw new IllegalReconnectException();
+    }
+    Assert.assertNull(getConnection());
+    wireNewConnection(newConnection);
+    logger.debug("reconnect connection attach {} {}", this.getConnectionID().getClientID(), getConnection());
   }
 
   @Override
@@ -129,16 +133,6 @@ private boolean verifyAck(WireProtocolMessage message) {
     return message instanceof TransportHandshakeMessage && ((TransportHandshakeMessage) message).isAck();
   }
 
-  private final class RestartConnectionAttacher implements ConnectionAttacher {
-
-    @Override
-    public void attachNewConnection(TCConnectionEvent closeEvent, TCConnection oldConnection, TCConnection newConnection) {
-      wireNewConnection(newConnection);
-      log("Attaching new connection to transport: " + newConnection);
-    }
-
-  }
-
   @Override
   public String toString() {
     return "ServerMessageTransport{connection=" + getConnection() + '}';

common/src/test/java/com/tc/net/protocol/transport/ServerMessageTransportTest.java

@@ -78,18 +78,16 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
         return null;
       }
     }).when(connection).addListener(ArgumentMatchers.any(TCConnectionEventListener.class));
+    when(connection.isConnected()).thenReturn(Boolean.TRUE);
     TransportHandshakeErrorHandler errHdr = mock(TransportHandshakeErrorHandler.class);
     TransportHandshakeMessageFactory factory = mock(TransportHandshakeMessageFactory.class);
     ServerMessageTransport transport = new ServerMessageTransport(connection, errHdr, factory);
     transport.initConnectionID(id);
     transport.addTransportListener(checker);
 
-    Assert.assertTrue(transport.status.isStartOpen());
+    Assert.assertTrue(transport.status.isConnected());
 
     TCConnectionEvent event = new TCConnectionEvent(connection);
-    transport.connectEvent(event);
-
-    Assert.assertTrue(transport.status.isConnected());
 
     for (TCConnectionEventListener trigger : listeners) {
       trigger.closeEvent(event);