TransportSet implements WithLogId { final ClientTransport obtainActiveTransport() { ClientTransport savedTransport = activeTransport; if (savedTransport != null) { return savedTransport; } synchronized (lock) { if (activeTransport == null) { if (shutdown) { return SHUTDOWN_TRANSPORT; } DelayedClientTransport delayedTransport = new DelayedClientTransport(appExecutor); transports.add(delayedTransport); delayedTransport.start(new BaseTransportListener(delayedTransport)); activeTransport = delayedTransport; scheduleConnection(delayedTransport); } return activeTransport; } } TransportSet(EquivalentAddressGroup addressGroup, String authority,
      LoadBalancer<ClientTransport> loadBalancer, BackoffPolicy.Provider backoffPolicyProvider,
      ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
      Executor appExecutor, Callback callback); @VisibleForTesting  TransportSet(EquivalentAddressGroup addressGroup, String authority,
      LoadBalancer<ClientTransport> loadBalancer, BackoffPolicy.Provider backoffPolicyProvider,
      ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
      Executor appExecutor, Callback callback, Stopwatch backoffWatch); @Override String getLogId();  }
@Test public void singleAddressReconnect() { SocketAddress addr = mock(SocketAddress.class); createTransportSet(addr); int transportsCreated = 0; int backoff1Consulted = 0; int backoff2Consulted = 0; int backoffReset = 0; int onAllAddressesFailed = 0; transportSet.obtainActiveTransport(); verify(mockBackoffPolicyProvider, times(++backoffReset)).get(); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); verify(mockTransportSetCallback, times(++onAllAddressesFailed)).onAllAddressesFailed(); transportSet.obtainActiveTransport(); verify(mockBackoffPolicy1, times(++backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); fakeClock.forwardMillis(9); verify(mockTransportFactory, times(transportsCreated)).newClientTransport(addr, authority); fakeClock.forwardMillis(1); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); verify(mockTransportSetCallback, times(++onAllAddressesFailed)).onAllAddressesFailed(); transportSet.obtainActiveTransport(); verify(mockBackoffPolicy1, times(++backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); fakeClock.forwardMillis(99); verify(mockTransportFactory, times(transportsCreated)).newClientTransport(addr, authority); fakeClock.forwardMillis(1); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); transports.peek().listener.transportReady(); verify(mockTransportSetCallback, never()).onConnectionClosedByServer(any(Status.class)); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); verify(mockTransportSetCallback).onConnectionClosedByServer(same(Status.UNAVAILABLE)); verify(mockTransportSetCallback, times(onAllAddressesFailed)).onAllAddressesFailed(); transportSet.obtainActiveTransport(); verify(mockBackoffPolicyProvider, times(++backoffReset)).get(); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); verify(mockBackoffPolicy1, times(backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicy2, times(backoff2Consulted)).nextBackoffMillis(); verifyNoMoreInteractions(mockTransportSetCallback); }
@Test public void twoAddressesReconnect() { SocketAddress addr1 = mock(SocketAddress.class); SocketAddress addr2 = mock(SocketAddress.class); createTransportSet(addr1, addr2); int transportsAddr1 = 0; int transportsAddr2 = 0; int backoff1Consulted = 0; int backoff2Consulted = 0; int backoff3Consulted = 0; int backoffReset = 0; int onAllAddressesFailed = 0; DelayedClientTransport delayedTransport1 = (DelayedClientTransport) transportSet.obtainActiveTransport(); verify(mockBackoffPolicyProvider, times(++backoffReset)).get(); verify(mockTransportFactory, times(++transportsAddr1)).newClientTransport(addr1, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); assertNull(delayedTransport1.getTransportSupplier()); verify(mockTransportSetCallback, times(onAllAddressesFailed)).onAllAddressesFailed(); DelayedClientTransport delayedTransport2 = (DelayedClientTransport) transportSet.obtainActiveTransport(); assertSame(delayedTransport1, delayedTransport2); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); verify(mockTransportFactory, times(++transportsAddr2)).newClientTransport(addr2, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); assertTrue(delayedTransport2.getTransportSupplier().get() instanceof FailingClientTransport); verify(mockTransportSetCallback, times(++onAllAddressesFailed)).onAllAddressesFailed(); DelayedClientTransport delayedTransport3 = (DelayedClientTransport) transportSet.obtainActiveTransport(); assertNotSame(delayedTransport2, delayedTransport3); verify(mockBackoffPolicy1, times(++backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); fakeClock.forwardMillis(9); verify(mockTransportFactory, times(transportsAddr1)).newClientTransport(addr1, authority); fakeClock.forwardMillis(1); verify(mockTransportFactory, times(++transportsAddr1)).newClientTransport(addr1, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); assertNull(delayedTransport3.getTransportSupplier()); verify(mockTransportSetCallback, times(onAllAddressesFailed)).onAllAddressesFailed(); DelayedClientTransport delayedTransport4 = (DelayedClientTransport) transportSet.obtainActiveTransport(); assertSame(delayedTransport3, delayedTransport4); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); verify(mockTransportFactory, times(++transportsAddr2)).newClientTransport(addr2, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); assertTrue(delayedTransport4.getTransportSupplier().get() instanceof FailingClientTransport); verify(mockTransportSetCallback, times(++onAllAddressesFailed)).onAllAddressesFailed(); DelayedClientTransport delayedTransport5 = (DelayedClientTransport) transportSet.obtainActiveTransport(); assertNotSame(delayedTransport4, delayedTransport5); verify(mockBackoffPolicy1, times(++backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicyProvider, times(backoffReset)).get(); fakeClock.forwardMillis(99); verify(mockTransportFactory, times(transportsAddr1)).newClientTransport(addr1, authority); fakeClock.forwardMillis(1); verify(mockTransportFactory, times(++transportsAddr1)).newClientTransport(addr1, authority); transports.peek().listener.transportReady(); assertSame(transports.peek().transport, delayedTransport5.getTransportSupplier().get()); verify(mockTransportSetCallback, never()).onConnectionClosedByServer(any(Status.class)); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); verify(mockTransportSetCallback).onConnectionClosedByServer(same(Status.UNAVAILABLE)); verify(mockTransportSetCallback, times(onAllAddressesFailed)).onAllAddressesFailed(); DelayedClientTransport delayedTransport6 = (DelayedClientTransport) transportSet.obtainActiveTransport(); assertNotSame(delayedTransport5, delayedTransport6); verify(mockBackoffPolicyProvider, times(++backoffReset)).get(); verify(mockTransportFactory, times(++transportsAddr1)).newClientTransport(addr1, authority); verify(mockBackoffPolicy1, times(backoff1Consulted)).nextBackoffMillis(); verify(mockBackoffPolicy2, times(backoff2Consulted)).nextBackoffMillis(); verify(mockBackoffPolicy3, times(backoff3Consulted)).nextBackoffMillis(); verifyNoMoreInteractions(mockTransportSetCallback); }
@Test public void connectIsLazy() { SocketAddress addr = mock(SocketAddress.class); createTransportSet(addr); int transportsCreated = 0; verify(mockTransportFactory, times(transportsCreated)).newClientTransport(addr, authority); assertNotNull(transportSet.obtainActiveTransport()); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); fakeClock.forwardMillis(10); verify(mockTransportFactory, times(transportsCreated)).newClientTransport(addr, authority); assertNotNull(transportSet.obtainActiveTransport()); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); transports.poll().listener.transportShutdown(Status.UNAVAILABLE); assertNotNull(transportSet.obtainActiveTransport()); verify(mockTransportFactory, times(transportsCreated)).newClientTransport(addr, authority); fakeClock.forwardMillis(100); verify(mockTransportFactory, times(++transportsCreated)).newClientTransport(addr, authority); }