HonoConnectionImpl implements HonoConnection { @Override public final Future<HonoConnection> connect() { return connect(null); } HonoConnectionImpl(final Vertx vertx, final ClientConfigProperties clientConfigProperties);  HonoConnectionImpl(final Vertx vertx, final ConnectionFactory connectionFactory,
            final ClientConfigProperties clientConfigProperties); @Override final Vertx getVertx(); @Autowired(required = false) final void setTracer(final Tracer opentracingTracer); @Override final Tracer getTracer(); @Override final ClientConfigProperties getConfig(); @Override final void addDisconnectListener(final DisconnectListener<HonoConnection> listener); @Override final void addReconnectListener(final ReconnectListener<HonoConnection> listener); @Override final Future<T> executeOnContext(final Handler<Promise<T>> codeToRun); @Override final Future<Void> isConnected(); @Override final Future<Void> isConnected(final long waitForCurrentConnectAttemptTimeout); @Override final boolean isShutdown(); @Override final boolean supportsCapability(final Symbol capability); @Override final Future<HonoConnection> connect(); @Override final Future<HonoConnection> connect(final ProtonClientOptions options); @Override void closeAndFree(
            final ProtonLink<?> link,
            final Handler<Void> closeHandler); @Override void closeAndFree(
            final ProtonLink<?> link,
            final long detachTimeOut,
            final Handler<Void> closeHandler); @Override final Future<ProtonSender> createSender(
            final String targetAddress,
            final ProtonQoS qos,
            final Handler<String> closeHook); @Override Future<ProtonReceiver> createReceiver(
            final String sourceAddress,
            final ProtonQoS qos,
            final ProtonMessageHandler messageHandler,
            final Handler<String> remoteCloseHook); @Override Future<ProtonReceiver> createReceiver(
            final String sourceAddress,
            final ProtonQoS qos,
            final ProtonMessageHandler messageHandler,
            final int preFetchSize,
            final Handler<String> remoteCloseHook); @Override Future<ProtonReceiver> createReceiver(
            final String sourceAddress,
            final ProtonQoS qos,
            final ProtonMessageHandler messageHandler,
            final int preFetchSize,
            final boolean autoAccept,
            final Handler<String> remoteCloseHook); @Override final void shutdown(); @Override final void shutdown(final Handler<AsyncResult<Void>> completionHandler); @Override final void disconnect(); @Override final void disconnect(final Handler<AsyncResult<Void>> completionHandler); @Override String getRemoteContainerId(); @Override String getContainerId();  }
@Test public void testConnectFailsAfterMaxConnectionAttempts(final VertxTestContext ctx) { props.setReconnectAttempts(2); props.setConnectTimeout(10); connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con) .setExpectedFailingConnectionAttempts(3); honoConnection = new HonoConnectionImpl(vertx, connectionFactory, props); honoConnection.connect().onComplete(ctx.failing(t -> { ctx.verify(() -> assertThat(((ServerErrorException) t).getErrorCode()).isEqualTo(HttpURLConnection.HTTP_UNAVAILABLE)); })); ctx.verify(() -> assertThat(connectionFactory.awaitFailure()).isTrue()); ctx.completeNow(); }
@Test public void testReconnectDelay(final VertxTestContext ctx) { final int reconnectAttempts = 5; props.setReconnectAttempts(reconnectAttempts); props.setReconnectMinDelay(10); props.setReconnectMaxDelay(1000); props.setReconnectDelayIncrement(100); props.setConnectTimeout(10); connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con) .setExpectedFailingConnectionAttempts(reconnectAttempts + 1); honoConnection = new HonoConnectionImpl(vertx, connectionFactory, props); honoConnection.connect().onComplete(ctx.failing(t -> { ctx.verify(() -> assertThat(((ServerErrorException) t).getErrorCode()).isEqualTo(HttpURLConnection.HTTP_UNAVAILABLE)); })); ctx.verify(() -> { assertThat(connectionFactory.awaitFailure()).isTrue(); final ArgumentCaptor<Long> delayValueCaptor = ArgumentCaptor.forClass(Long.class); verify(vertx, times(reconnectAttempts)).setTimer(delayValueCaptor.capture(), VertxMockSupport.anyHandler()); assertThat(delayValueCaptor.getAllValues().get(0)).isEqualTo(10L); }); ctx.completeNow(); }
@Test public void testConnectFailsWithClientErrorForTransientSaslSystemException(final VertxTestContext ctx) { props.setReconnectAttempts(2); props.setConnectTimeout(10); connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con) .setExpectedFailingConnectionAttempts(3) .failWith(new SaslSystemException(false, "SASL handshake failed due to a transient error")); honoConnection = new HonoConnectionImpl(vertx, connectionFactory, props); honoConnection.connect().onComplete(ctx.failing(t -> { ctx.verify(() -> assertThat(((ServiceInvocationException) t).getErrorCode()).isEqualTo(HttpURLConnection.HTTP_UNAVAILABLE)); })); ctx.verify(() -> assertThat(connectionFactory.awaitFailure()).isTrue()); ctx.completeNow(); }
@Test public void testDownstreamDisconnectTriggersReconnect(final VertxTestContext ctx) { connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con); props.setReconnectAttempts(1); final ProtonClientOptions options = new ProtonClientOptions() .setReconnectAttempts(0); honoConnection = new HonoConnectionImpl(vertx, connectionFactory, props); honoConnection.connect(options).onComplete(ctx.succeeding()); ctx.verify(() -> assertThat(connectionFactory.await()).isTrue()); connectionFactory.setExpectedSucceedingConnectionAttempts(1); connectionFactory.getDisconnectHandler().handle(con); ctx.verify(() -> assertThat(connectionFactory.await()).isTrue()); connectionFactory.setExpectedSucceedingConnectionAttempts(1); connectionFactory.getDisconnectHandler().handle(con); ctx.verify(() -> assertThat(connectionFactory.await()).isTrue()); ctx.completeNow(); }
@Test public void testConnectTriesToReconnectOnFailedConnectAttempt(final VertxTestContext ctx) { connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con) .setExpectedFailingConnectionAttempts(2); props.setReconnectAttempts(2); props.setConnectTimeout(10); honoConnection = new HonoConnectionImpl(vertx, connectionFactory, props); honoConnection.connect().onComplete(ctx.succeeding()); ctx.verify(() -> { assertThat(connectionFactory.awaitFailure()).isTrue(); assertThat(connectionFactory.await()).isTrue(); }); ctx.completeNow(); }