import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
private Future<MessageSender> getMessageSender() {

        final Future<MessageSender> result = Future.future();

        final Future<HonoClient> messagingConnectionTracker = Future.future();
        honoMessagingClient.connect(new ProtonClientOptions(), messagingConnectionTracker.completer());
        messagingConnectionTracker.compose(messagingClient -> {
            if (isEventMode()) {
                messagingClient.getOrCreateEventSender(HonoExampleConstants.TENANT_ID, result.completer());
            } else {
                messagingClient.getOrCreateTelemetrySender(HonoExampleConstants.TENANT_ID, result.completer());
            }
        }, result);

        return result;
    }
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
private void connectToDownstream(final ProtonClientOptions options, final Handler<AsyncResult<ProtonConnection>> connectResultHandler) {

        downstreamConnectionFactory.connect(
                options,
                this::onRemoteClose,
                this::onDisconnectFromDownstreamContainer,
                connectAttempt -> {
                    if (connectAttempt.succeeded()) {
                        this.downstreamConnection = connectAttempt.result();
                        metrics.incrementDownStreamConnections();
                        if (connectResultHandler != null) {
                            connectResultHandler.handle(Future.succeededFuture(connectAttempt.result()));
                        }
                    } else {
                        logger.info("failed to connect to downstream container: {}", connectAttempt.cause().getMessage());
                        if (retryOnFailedConnectAttempt) {
                            reconnect(connectResultHandler);
                        } else if (connectResultHandler != null) {
                            connectResultHandler.handle(Future.failedFuture(connectAttempt.cause()));
                        }
                    }
                });
    }
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
@Override
public void connect(
        final ProtonClientOptions options,
        final String username,
        final String password,
        final Handler<AsyncResult<ProtonConnection>> closeHandler,
        final Handler<ProtonConnection> disconnectHandler,
        final Handler<AsyncResult<ProtonConnection>> connectionResultHandler) {

    if (expectedConnectionAttempts.getCount() > 0) {
        expectedConnectionAttempts.countDown();
        this.disconnectHandler = disconnectHandler;
        this.closeHandler = closeHandler;
        if (connectionToCreate == null) {
            connectionResultHandler.handle(Future.failedFuture("cannot connect"));
        } else {
            connectionResultHandler.handle(Future.succeededFuture(connectionToCreate));
        }
    }
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that a request to create a request-response client is failed immediately when the
 * underlying connection to the server fails.
 * 
 * @param ctx The Vertx test context.
 */
@Test
public void testGetOrCreateRequestResponseClientFailsOnConnectionFailure(final TestContext ctx) {

    // GIVEN a client that tries to create a registration client for "tenant"
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con);
    final Async connected = ctx.async();
    final Async disconnected = ctx.async();
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.connect(new ProtonClientOptions(), ctx.asyncAssertSuccess(ok -> connected.complete()));
    connected.await(200);

    client.getOrCreateRequestResponseClient("registration/tenant", creationResultHandler -> {
        ctx.assertFalse(disconnected.isCompleted());
    }, ctx.asyncAssertFailure(cause -> {
        disconnected.complete();
    }));

    // WHEN the underlying connection fails
    connectionFactory.getDisconnectHandler().handle(con);

    // THEN all creation requests are failed
    disconnected.await(200);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that a request to create a message sender is failed immediately when the
 * underlying connection to the server fails.
 * 
 * @param ctx The Vertx test context.
 */
@Test
public void testGetOrCreateSenderFailsOnConnectionFailure(final TestContext ctx) {

    // GIVEN a client that tries to create a telemetry sender for "tenant"
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con);
    final Async connected = ctx.async();
    final Async disconnected = ctx.async();
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.connect(new ProtonClientOptions(), ctx.asyncAssertSuccess(ok -> connected.complete()));
    connected.await(200);

    client.getOrCreateSender("telemetry/tenant", creationResultHandler -> {
        ctx.assertFalse(disconnected.isCompleted());
    }, ctx.asyncAssertFailure(cause -> {
        disconnected.complete();
    }));

    // WHEN the underlying connection fails
    connectionFactory.getDisconnectHandler().handle(con);

    // THEN all creation requests are failed
    disconnected.await(200);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that a request to create a telemetry consumer is failed immediately when the
 * underlying connection to the server fails.
 * 
 * @param ctx The Vertx test context.
 */
@Test
public void testCreateTelemetryConsumerFailsOnConnectionFailure(final TestContext ctx) {

    // GIVEN a client that already tries to create a telemetry sender for "tenant"
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con);
    final Async connected = ctx.async();
    final Async disconnected = ctx.async();
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.connect(new ProtonClientOptions(), ctx.asyncAssertSuccess(ok -> connected.complete()));
    connected.await(200);

    client.createTelemetryConsumer("tenant", msg -> {}, ctx.asyncAssertFailure(cause -> {
        disconnected.complete();
    }));

    // WHEN the underlying connection fails
    connectionFactory.getDisconnectHandler().handle(con);

    // THEN all creation requests are failed
    disconnected.await(200);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that a request to create an event consumer is failed immediately when the
 * underlying connection to the server fails.
 * 
 * @param ctx The Vertx test context.
 */
@Test
public void testCreateEventConsumerFailsOnConnectionFailure(final TestContext ctx) {

    // GIVEN a client that already tries to create a telemetry sender for "tenant"
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con);
    final Async connected = ctx.async();
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.connect(new ProtonClientOptions(), ctx.asyncAssertSuccess(ok -> connected.complete()));
    connected.await(200);

    final Async disconnected = ctx.async();
    client.createEventConsumer("tenant", msg -> {}, ctx.asyncAssertFailure(cause -> {
        disconnected.complete();
    }));

    // WHEN the underlying connection fails
    connectionFactory.getDisconnectHandler().handle(con);

    // THEN all creation requests are failed
    disconnected.await(200);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the client tries to re-establish a lost connection to a server.
 * 
 * @param ctx The Vertx test context.
 */
@Test
public void testDownstreamDisconnectTriggersReconnect(final TestContext ctx) {

    final ProtonConnection connectionToCreate = mock(ProtonConnection.class);
    when(connectionToCreate.getRemoteContainer()).thenReturn("server");
    // expect the connection factory to be invoked twice
    // first on initial connection
    // second on re-connect attempt
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(connectionToCreate, 2);

    // GIVEN an client connected to a server
    final Async connected = ctx.async();
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.connect(new ProtonClientOptions().setReconnectAttempts(1), ctx.asyncAssertSuccess(ok -> connected.complete()));
    connected.await(200);

    // WHEN the downstream connection fails
    connectionFactory.getDisconnectHandler().handle(connectionToCreate);

    // THEN the adapter tries to reconnect to the downstream container
    connectionFactory.await(1, TimeUnit.SECONDS);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the client adapter repeatedly tries to connect until a connection is established.
 * 
 * @param ctx The test context.
 */
@Test
public void testConnectTriesToReconnectOnFailedConnectAttempt(final TestContext ctx) {

    // GIVEN a client that cannot connect to the server
    ProtonConnection con = mock(ProtonConnection.class);
    // expect the connection factory to fail twice and succeed on third connect attempt
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con, 1, 2);
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);

    // WHEN trying to connect
    Async disconnectHandlerInvocation = ctx.async();
    Async connectionEstablished = ctx.async();
    Handler<ProtonConnection> disconnectHandler = failedCon -> disconnectHandlerInvocation.complete();
    client.connect(
            new ProtonClientOptions().setReconnectAttempts(1),
            ctx.asyncAssertSuccess(ok -> connectionEstablished.complete()),
            disconnectHandler);

    // THEN the client repeatedly tries to connect
    connectionEstablished.await(4 * Constants.DEFAULT_RECONNECT_INTERVAL_MILLIS);
    // and sets the disconnect handler provided as a param in the connect method invocation
    connectionFactory.getDisconnectHandler().handle(con);
    disconnectHandlerInvocation.await(1000);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the client tries to re-connect to a server instance if the connection is closed by the peer.
 * 
 * @param ctx The test context.
 *
 */
@Test
public void testOnRemoteCloseTriggersReconnection(final TestContext ctx) {

    // GIVEN a client that is connected to a server
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con, 2);
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    Async disconnectHandlerInvocation = ctx.async();
    Handler<ProtonConnection> disconnectHandler = failedCon -> disconnectHandlerInvocation.complete();
    client.connect(new ProtonClientOptions().setReconnectAttempts(1), attempt -> {}, disconnectHandler);

    // WHEN the peer closes the connection
    connectionFactory.getCloseHandler().handle(Future.failedFuture("shutting down for maintenance"));

    // THEN the client invokes the disconnect handler provided in the original connect method call
    disconnectHandlerInvocation.await(500);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that it fails to connect after client was shutdown.
 * 
 * @param ctx The test context.
 *
 */
@Test
public void testConnectFailsAfterShutdown(final TestContext ctx) {

    // GIVEN a shutdown client
    ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    client.shutdown();

    // WHEN client connects
    Async connectionHandlerInvocation = ctx.async();
    Handler<AsyncResult<HonoClient>> connectionHandler = result -> {
        if (result.failed()) {
            connectionHandlerInvocation.complete();
        }
    };
    client.connect(new ProtonClientOptions(), connectionHandler );

    //THEN connect fails
    connectionHandlerInvocation.await(500);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the client does not try to re-connect to a server instance if the client was shutdown.
 * 
 * @param ctx The test context.
 *
 */
@Test
public void testDoesNotTriggerReconnectionAfterShutdown(final TestContext ctx) {

    // GIVEN a client that tries to connect to a server but does not succeed
    ProtonConnection con = mock(ProtonConnection.class);
    DisconnectHandlerProvidingConnectionFactory connectionFactory = new DisconnectHandlerProvidingConnectionFactory(con, 1, Integer.MAX_VALUE);
    HonoClientImpl client = new HonoClientImpl(vertx, connectionFactory);
    Async connectionHandlerInvocation = ctx.async();
    Handler<AsyncResult<HonoClient>> connectionHandler = result -> {
        if (result.failed()) {
            connectionHandlerInvocation.complete();
        }
    };
    client.connect(new ProtonClientOptions().setReconnectAttempts(1), connectionHandler);

    // WHEN client gets shutdown
    client.shutdown();

    // THEN reconnect gets stopped, i.e. connection fails
    connectionHandlerInvocation.await(1000);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
@Override
public void connect(
        final ProtonClientOptions options,
        final String username,
        final String password,
        final Handler<AsyncResult<ProtonConnection>> closeHandler,
        final Handler<ProtonConnection> disconnectHandler,
        final Handler<AsyncResult<ProtonConnection>> connectionResultHandler) {

    this.closeHandler = closeHandler;
    this.disconnectHandler = disconnectHandler;
    if (expectedFailingConnectionAttempts.getCount() > 0) {
        expectedFailingConnectionAttempts.countDown();
        connectionResultHandler.handle(Future.failedFuture("cannot connect"));
    } else {
        expectedSucceedingConnectionAttemps.countDown();
        connectionResultHandler.handle(Future.succeededFuture(connectionToCreate));
    }
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that an MQTT server is bound to the insecure port during startup and connections
 * to required services have been established.
 * 
 * @param ctx The helper to use for running async tests on vertx.
 */
@SuppressWarnings("unchecked")
@Test
public void testStartup(final TestContext ctx) {

    MqttServer server = getMqttServer(false);
    AbstractVertxBasedMqttProtocolAdapter<ProtocolAdapterProperties> adapter = getAdapter(server);

    Async startup = ctx.async();

    Future<Void> startupTracker = Future.future();
    startupTracker.setHandler(ctx.asyncAssertSuccess(s -> {
        startup.complete();
    }));
    adapter.start(startupTracker);

    startup.await();

    verify(server).listen(any(Handler.class));
    verify(server).endpointHandler(any(Handler.class));
    verify(messagingClient).connect(any(ProtonClientOptions.class), any(Handler.class), any(Handler.class));
    verify(registrationClient).connect(any(ProtonClientOptions.class), any(Handler.class), any(Handler.class));
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that a client provided http server is started instead of creating and starting a new http server.
 * 
 * @param ctx The helper to use for running async tests on vertx.
 * @throws Exception if the test fails.
 */
@SuppressWarnings("unchecked")
@Test
public void testStartUsesClientProvidedHttpServer(final TestContext ctx) throws Exception {

    // GIVEN an adapter with a client provided http server
    HttpServer server = getHttpServer(false);
    AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null);
    adapter.setCredentialsAuthProvider(credentialsAuthProvider);

    // WHEN starting the adapter
    Async startup = ctx.async();
    Future<Void> startupTracker = Future.future();
    startupTracker.setHandler(ctx.asyncAssertSuccess(s -> {
        startup.complete();
    }));
    adapter.start(startupTracker);

    // THEN the client provided http server has been configured and started
    startup.await();
    verify(server).requestHandler(any(Handler.class));
    verify(server).listen(any(Handler.class));
    verify(messagingClient).connect(any(ProtonClientOptions.class), any(Handler.class), any(Handler.class));
    verify(registrationClient).connect(any(ProtonClientOptions.class), any(Handler.class), any(Handler.class));
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
private void addTlsTrustOptions(final ProtonClientOptions clientOptions) {

        if (config.isTlsEnabled()) {
            clientOptions.setSsl(true);
        }

        if (clientOptions.getTrustOptions() == null) {
            TrustOptions trustOptions = config.getTrustOptions();
            if (trustOptions != null) {
                clientOptions.setSsl(true).setTrustOptions(trustOptions);
            }
        }

        if (clientOptions.isSsl()) {
            if (config.isHostnameVerificationRequired()) {
                clientOptions.setHostnameVerificationAlgorithm("HTTPS");
            } else {
                clientOptions.setHostnameVerificationAlgorithm("");
            }
        }
    }
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the given result handler is invoked if a connection attempt fails.
 * 
 * @param ctx The vert.x test context.
 */
@Test
public void testConnectInvokesHandlerOnfailureToConnect(final TestContext ctx) {

    // GIVEN a factory configured to connect to a non-existing server
    ConnectionFactoryImpl factory = new ConnectionFactoryImpl(vertx, props);

    // WHEN trying to connect to the server
    final Async handlerInvocation = ctx.async();

    ProtonClientOptions options = new ProtonClientOptions().setConnectTimeout(100);
    factory.connect(options, null, null, ctx.asyncAssertFailure(t -> {
        handlerInvocation.complete();
    }));

    // THEN the connection attempt fails and the given handler is invoked
    handlerInvocation.await(2000);
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the factory does not enable SASL_PLAIN if the username and password are empty
 * strings.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testConnectDoesNotUseSaslPlainForEmptyUsernameAndPassword(final TestContext ctx) {

    // GIVEN a factory configured to connect to a server
    final ProtonClientOptions options = new ProtonClientOptions();
    final ProtonClient client = mock(ProtonClient.class);
    final ConnectionFactoryImpl factory = new ConnectionFactoryImpl(vertx, props);
    factory.setProtonClient(client);

    // WHEN connecting to the server using empty strings for username and password
    factory.connect(options, "", "", null, null, c -> {});

    // THEN the factory does not enable the SASL_PLAIN mechanism when establishing
    // the connection
    ArgumentCaptor<ProtonClientOptions> optionsCaptor = ArgumentCaptor.forClass(ProtonClientOptions.class);
    verify(client).connect(optionsCaptor.capture(), anyString(), anyInt(), eq(""), eq(""), any(Handler.class));
    assertFalse(optionsCaptor.getValue().getEnabledSaslMechanisms().contains("PLAIN"));
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the factory enables SASL_PLAIN if the username and password are non-empty
 * strings.
 * 
 * @param ctx The vert.x test context.
 */
@SuppressWarnings("unchecked")
@Test
public void testConnectAddsSaslPlainForNonEmptyUsernameAndPassword(final TestContext ctx) {

    // GIVEN a factory configured to connect to a server
    final ProtonClientOptions options = new ProtonClientOptions();
    final ProtonClient client = mock(ProtonClient.class);
    final ConnectionFactoryImpl factory = new ConnectionFactoryImpl(vertx, props);
    factory.setProtonClient(client);

    // WHEN connecting to the server using non-empty strings for username and password
    factory.connect(options, "user", "pw", null, null, c -> {});

    // THEN the factory uses SASL_PLAIN when establishing the connection
    ArgumentCaptor<ProtonClientOptions> optionsCaptor = ArgumentCaptor.forClass(ProtonClientOptions.class);
    verify(client).connect(optionsCaptor.capture(), anyString(), anyInt(), eq("user"), eq("pw"), any(Handler.class));
    assertTrue(optionsCaptor.getValue().getEnabledSaslMechanisms().contains("PLAIN"));
}
 

import io.vertx.proton.ProtonClientOptions; //导入依赖的package包/类
/**
 * Verifies that the factory uses TLS when connecting to the peer if no trust store
 * is configured but TLS has been enabled explicitly.
 */
@SuppressWarnings("unchecked")
@Test
public void testConnectEnablesSslIfExplicitlyConfigured() {

    // GIVEN a factory configured to connect to a server using TLS
    final ClientConfigProperties config = new ClientConfigProperties();
    config.setHost("remote.host");
    config.setTlsEnabled(true);
    final ProtonClient client = mock(ProtonClient.class);
    final ConnectionFactoryImpl factory = new ConnectionFactoryImpl(vertx, config);
    factory.setProtonClient(client);

    // WHEN connecting to the server
    factory.connect(null, null, null, c -> {});

    // THEN the factory uses TLS when establishing the connection
    ArgumentCaptor<ProtonClientOptions> optionsCaptor = ArgumentCaptor.forClass(ProtonClientOptions.class);
    verify(client).connect(optionsCaptor.capture(), eq("remote.host"), anyInt(), anyString(), anyString(), any(Handler.class));
    assertTrue(optionsCaptor.getValue().isSsl());
}