Merge pull request #1332 from myronkscott/fix_reconnect2

fixes to prevent client reconnection outside the server's 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/main/java/com/tc/net/protocol/transport/ServerStackProvider.java

@@ -54,7 +54,7 @@
 public class ServerStackProvider implements NetworkStackProvider, MessageTransportListener, ProtocolAdaptorFactory {
   private static final Logger logger = LoggerFactory.getLogger(ServerStackProvider.class);
 
-  private final Map<ClientID, ServerNetworkStackHarness> harnesses          = new ConcurrentHashMap<ClientID, ServerNetworkStackHarness>();
+  private final Map<ClientID, ServerNetworkStackHarness> harnesses          = new ConcurrentHashMap<>();
   private final NetworkStackHarnessFactory       harnessFactory;
   private final ServerMessageChannelFactory      channelFactory;
   private final TransportHandshakeMessageFactory handshakeMessageFactory;
@@ -63,7 +63,7 @@ public class ServerStackProvider implements NetworkStackProvider, MessageTranspo
   private final WireProtocolAdaptorFactory       wireProtocolAdaptorFactory;
   private final WireProtocolMessageSink          wireProtoMsgsink;
   private final MessageTransportFactory          messageTransportFactory;
-  private final List<MessageTransportListener>   transportListeners = new ArrayList<MessageTransportListener>();
+  private final List<MessageTransportListener>   transportListeners = new ArrayList<>();
   private final ReentrantLock                    licenseLock;
   private final RedirectAddressProvider                 activeProvider;
   private final Predicate<MessageTransport>                validateTransport;
@@ -213,7 +213,7 @@ public void notifyTransportClosed(MessageTransport transport) {
       if (!connectionId.isJvmIDNull()) {
         boolean removed = this.connectionPolicy.clientDisconnected(connectionId);
         if (removed) {
-          logger.warn("connectionid not removed be transport disconnect");
+          logger.warn("connectionid not removed by transport disconnect");
         }
       }
     }

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

@@ -21,6 +21,7 @@
 import com.tc.net.core.TCConnection;
 import com.tc.net.core.event.TCConnectionEvent;
 import com.tc.net.core.event.TCConnectionEventListener;
+import com.tc.net.protocol.IllegalReconnectException;
 import com.tc.util.Assert;
 import java.util.ArrayList;
 import java.util.List;
@@ -78,18 +79,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);
@@ -126,4 +125,47 @@ public Void answer(InvocationOnMock invocation) throws Throwable {
     verify(checker, times(2)).notifyTransportDisconnected(eq(transport), eq(false));
     verify(checker, times(1)).notifyTransportDisconnected(eq(transport), eq(true));
   }
+  
+  @Test
+  public void testDoubleAttachFails() throws Exception {
+    ConnectionID id = new ConnectionID("JVM", 1);
+    TCConnection connection = mock(TCConnection.class);
+    
+    MessageTransportListener checker = mock(MessageTransportListener.class);
+    
+    final List<TCConnectionEventListener> listeners = new ArrayList<TCConnectionEventListener>();
+    doAnswer(new Answer<Void>() {
+      @Override
+      public Void answer(InvocationOnMock invocation) throws Throwable {
+        listeners.add((TCConnectionEventListener)invocation.getArguments()[0]);
+        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(errHdr, factory);
+    transport.initConnectionID(id);
+    transport.addTransportListener(checker);
+
+    Assert.assertFalse(transport.status.isConnected());
+    
+    transport.attachNewConnection(connection);
+    Assert.assertTrue(transport.status.isConnected());
+        
+    try {
+      transport.attachNewConnection(connection);
+      Assert.fail();
+    } catch (IllegalReconnectException illegal) {
+      //  expected
+    }
+    transport.close();
+    Assert.assertFalse(transport.status.isConnected());
+    try {
+      transport.attachNewConnection(connection);
+      Assert.fail();
+    } catch (IllegalReconnectException illegal) {
+      //  expected
+    }
+  }
 }