TCPConduit implements Runnable { public Connection getConnection(InternalDistributedMember memberAddress, final boolean preserveOrder, boolean retry, long startTime, long ackTimeout, long ackSATimeout) throws IOException, DistributedSystemDisconnectedException { if (stopped) { throw new DistributedSystemDisconnectedException("The conduit is stopped"); } InternalDistributedMember memberInTrouble = null; Connection conn = null; for (boolean breakLoop = false;;) { stopper.checkCancelInProgress(null); boolean interrupted = Thread.interrupted(); try { if (retry || conn != null) { if (!membership.memberExists(memberAddress) || membership.isShunned(memberAddress) || membership.shutdownInProgress()) { throw new IOException("TCP/IP connection lost and member is not in view"); } try { Thread.sleep(100); } catch (InterruptedException e) { interrupted = true; stopper.checkCancelInProgress(e); } if (!membership.memberExists(memberAddress) || membership.isShunned(memberAddress)) { throw new IOException("TCP/IP connection lost and member is not in view"); } if (memberInTrouble == null) { memberInTrouble = memberAddress; logger.warn("Attempting TCP/IP reconnect to {}", memberInTrouble); } else { if (logger.isDebugEnabled()) { logger.debug("Attempting TCP/IP reconnect to {}", memberInTrouble); } } getStats().incReconnectAttempts(); if (conn != null) { try { if (logger.isDebugEnabled()) { logger.debug("Closing old connection. conn={} before retrying. memberInTrouble={}", conn, memberInTrouble); } conn.closeForReconnect("closing before retrying"); } catch (CancelException ex) { throw ex; } catch (Exception ex) { } } } Exception problem = null; try { boolean retryForOldConnection; boolean debugRetry = false; do { retryForOldConnection = false; conn = getConTable().get(memberAddress, preserveOrder, startTime, ackTimeout, ackSATimeout); if (conn == null) { problem = new IOException( String.format("Unable to reconnect to server; possible shutdown: %s", memberAddress)); } else if (conn.isClosing() || !conn.getRemoteAddress().equals(memberAddress)) { if (logger.isDebugEnabled()) { logger.debug("Got an old connection for {}: {}@{}", memberAddress, conn, conn.hashCode()); } conn.closeOldConnection("closing old connection"); conn = null; retryForOldConnection = true; debugRetry = true; } } while (retryForOldConnection); if (debugRetry && logger.isDebugEnabled()) { logger.debug("Done removing old connections"); } } catch (ConnectionException e) { problem = e; breakLoop = true; } catch (IOException e) { problem = e; if (AlertingAction.isThreadAlerting()) { if (logger.isDebugEnabled()) { logger.debug("Giving up connecting to alert listener {}", memberAddress); } breakLoop = true; } } if (problem != null) { if (!membership.memberExists(memberAddress) || membership.isShunned(memberAddress)) { if (memberInTrouble != null) { logger.info("Ending reconnect attempt because {} has disappeared.", memberInTrouble); } throw new IOException( String.format("Peer has disappeared from view: %s", memberAddress)); } if (membership.shutdownInProgress()) { if (memberInTrouble != null) { logger.info("Ending reconnect attempt to {} because shutdown has started.", memberInTrouble); } stopper.checkCancelInProgress(null); throw new DistributedSystemDisconnectedException( "Abandoned because shutdown is in progress"); } if (memberInTrouble == null) { logger.warn("Error sending message to {} (will reattempt): {}", memberAddress, problem); memberInTrouble = memberAddress; } else { if (logger.isDebugEnabled()) { logger.debug("Error sending message to {}", memberAddress, problem); } } if (breakLoop) { if (problem instanceof IOException) { if (problem.getMessage().startsWith("Cannot form connection to alert listener")) { throw new AlertingIOException((IOException) problem); } throw (IOException) problem; } throw new IOException( String.format("Problem connecting to %s", memberAddress), problem); } continue; } if (memberInTrouble != null) { logger.info("Successfully reconnected to member {}", memberInTrouble); if (logger.isTraceEnabled()) { logger.trace("new connection is {} memberAddress={}", conn, memberAddress); } } return conn; } finally { if (interrupted) { Thread.currentThread().interrupt(); } } } } TCPConduit(Membership mgr, int port, InetAddress address, boolean isBindAddress,
      DirectChannel receiver, Properties props); @VisibleForTesting  TCPConduit(Membership mgr, int port, InetAddress address, boolean isBindAddress,
      DirectChannel receiver, Properties props,
      Function<TCPConduit, ConnectionTable> connectionTableFactory, SocketCreator socketCreator,
      Runnable localHostValidation, boolean startAcceptor); static void init(); Membership<InternalDistributedMember> getMembership(); static int getBackLog(); void emergencyClose(); void stop(Exception cause); void restart(); @Override void run(); void getThreadOwnedOrderedConnectionState(DistributedMember member, Map result); void waitForThreadOwnedOrderedConnectionState(DistributedMember member, Map channelState); InetSocketAddress getSocketId(); int getPort(); InternalDistributedMember getMemberId(); void setMemberId(InternalDistributedMember addr); DistributionConfig getConfig(); Connection getConnection(InternalDistributedMember memberAddress,
      final boolean preserveOrder, boolean retry, long startTime, long ackTimeout,
      long ackSATimeout); @Override String toString(); DistributionManager getDM(); void removeEndpoint(DistributedMember mbr, String reason, boolean notifyDisconnect); boolean hasReceiversFor(DistributedMember endPoint); DMStats getStats(); boolean useSSL(); BufferPool getBufferPool(); CancelCriterion getCancelCriterion();  }
@Test public void getConnectionThrowsAlertingIOException_ifCaughtIOException_whileAlerting() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); doThrow(new IOException("Cannot form connection to alert listener")) .when(connectionTable).get(eq(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(eq(member))) .thenReturn(true); when(membership.isShunned(same(member))) .thenReturn(false); AlertingAction.execute(() -> { Throwable thrown = catchThrowable(() -> { tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); }); assertThat(thrown) .isInstanceOf(AlertingIOException.class); }); }
@Test public void getConnectionRethrows_ifCaughtIOException_whileNotAlerting() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); Connection connection = mock(Connection.class); when(connection.getRemoteAddress()) .thenReturn(member); doThrow(new IOException("Cannot form connection to alert listener")) .doReturn(connection) .when(connectionTable).get(eq(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(eq(member))) .thenReturn(true); when(membership.isShunned(same(member))) .thenReturn(false); Connection value = tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); assertThat(value) .isSameAs(connection); }
@Test public void getConnectionRethrows_ifCaughtIOException_whenMemberDoesNotExist() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); doThrow(new IOException("Cannot form connection to alert listener")) .when(connectionTable).get(eq(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(eq(member))) .thenReturn(false); Throwable thrown = catchThrowable(() -> { tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); }); assertThat(thrown) .isInstanceOf(IOException.class) .isNotInstanceOf(AlertingIOException.class); }
@Test public void getConnectionRethrows_ifCaughtIOException_whenMemberIsShunned() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); doThrow(new IOException("Cannot form connection to alert listener")) .when(connectionTable).get(same(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(same(member))) .thenReturn(true); when(membership.isShunned(same(member))) .thenReturn(true); Throwable thrown = catchThrowable(() -> { tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); }); assertThat(thrown) .isInstanceOf(IOException.class) .isNotInstanceOf(AlertingIOException.class); }
@Test public void getConnectionThrowsDistributedSystemDisconnectedException_ifCaughtIOException_whenShutdownIsInProgress() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); doThrow(new IOException("Cannot form connection to alert listener")) .when(connectionTable).get(same(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(same(member))) .thenReturn(true); when(membership.isShunned(same(member))) .thenReturn(false); when(membership.shutdownInProgress()) .thenReturn(true); Throwable thrown = catchThrowable(() -> { tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); }); assertThat(thrown) .isInstanceOf(DistributedSystemDisconnectedException.class) .hasMessage("Abandoned because shutdown is in progress"); }
@Test public void getConnectionThrowsDistributedSystemDisconnectedException_ifCaughtIOException_whenShutdownIsInProgress_andCancelIsInProgress() throws Exception { TCPConduit tcpConduit = new TCPConduit(membership, 0, localHost, false, directChannel, new Properties(), TCPConduit -> connectionTable, socketCreator, doNothing(), false); InternalDistributedMember member = mock(InternalDistributedMember.class); doThrow(new IOException("Cannot form connection to alert listener")) .when(connectionTable).get(same(member), anyBoolean(), anyLong(), anyLong(), anyLong()); when(membership.memberExists(same(member))) .thenReturn(true); when(membership.isShunned(same(member))) .thenReturn(false); when(membership.shutdownInProgress()) .thenReturn(true); Throwable thrown = catchThrowable(() -> { tcpConduit.getConnection(member, false, false, 0L, 0L, 0L); }); assertThat(thrown) .isInstanceOf(DistributedSystemDisconnectedException.class) .hasMessage("Abandoned because shutdown is in progress"); }