private void mapClientOptionsBase(AmqpProperties from, AmqpClientOptions to) {
    to.setConnectTimeout(from.getConnectTimeout());
    to.setTrustAll(from.isTrustAll());
    to.setMetricsName(from.getMetricsName());
    to.setLocalAddress(from.getLocalAddress());

    if (from.getProxy().isEnabled()) {
        ProxyOptions proxyOptions = new ProxyOptions()
            .setHost(from.getProxy().getHost())
            .setPort(from.getProxy().getPort())
            .setUsername(from.getProxy().getUsername())
            .setPassword(from.getProxy().getPassword())
            .setType(ProxyType.valueOf(from.getProxy().getType().name()));

        to.setProxyOptions(proxyOptions);
    }
}
 

@Produces
@Named("my-named-options")
public AmqpClientOptions getNamedOptions() {
    // You can use the produced options to configure the TLS connection
    PemKeyCertOptions keycert = new PemKeyCertOptions()
        .addCertPath("./tls/tls.crt")
        .addKeyPath("./tls/tls.key");
    PemTrustOptions trust =
        new PemTrustOptions().addCertPath("./tlc/ca.crt");

    return new AmqpClientOptions()
        .setSsl(true)
        .setPemKeyCertOptions(keycert)
        .setPemTrustOptions(trust)
        .addEnabledSaslMechanism("EXTERNAL")
        .setHostnameVerificationAlgorithm("")
        .setConnectTimeout(30000)
        .setReconnectInterval(5000)
        .setContainerId("my-container");
}
 

@Override
public Completable rxStart() {
  updateProducer = KafkaProducer.create(vertx, kafkaConfig());

  AmqpClientOptions amqpOptions = amqpConfig();
  AmqpReceiverOptions receiverOptions = new AmqpReceiverOptions()
    .setAutoAcknowledgement(false)
    .setDurable(true);

  AmqpClient.create(vertx, amqpOptions)
    .rxConnect()
    .flatMap(conn -> conn.rxCreateReceiver("step-events", receiverOptions))
    .flatMapPublisher(AmqpReceiver::toFlowable)
    .doOnError(this::logAmqpError)
    .retryWhen(this::retryLater)
    .subscribe(this::handleAmqpMessage);

  Router router = Router.router(vertx);
  router.post().handler(BodyHandler.create());
  router.post("/ingest").handler(this::httpIngest);

  return vertx.createHttpServer()
    .requestHandler(router)
    .rxListen(HTTP_PORT)
    .ignoreElement();
}
 

@Bean
public AmqpClient amqpClient(Vertx vertx, AmqpProperties properties) {
    AmqpPropertiesConverter propertiesConverter = new AmqpPropertiesConverter();
    AmqpClientOptions options = propertiesConverter.toAmqpClientOptions(properties);

    return new SnowdropAmqpClient(getAxleAmqpClient(vertx, options), new MessageConverter());
}
 

AmqpClientOptions toAmqpClientOptions(AmqpProperties from) {
    AmqpClientOptions to = new AmqpClientOptions();

    mapAmqpClientOptions(from, to);
    mapProtonClientOptions(from, to);
    mapNetClientOptions(from, to);
    mapClientOptionsBase(from, to);
    mapTcpSslOptions(from, to);
    mapNetworkOptions(from, to);

    return to;
}
 

private void mapAmqpClientOptions(AmqpProperties from, AmqpClientOptions to) {
    to.setHost(from.getHost());
    to.setPort(from.getPort());
    to.setUsername(from.getUsername());
    to.setPassword(from.getPassword());
    to.setContainerId(from.getContainerId());
}
 

private void mapProtonClientOptions(AmqpProperties from, AmqpClientOptions to) {
    from.getEnabledSaslMechanisms().forEach(to::addEnabledSaslMechanism);
    to.setHeartbeat(from.getHeartbeat());
    to.setMaxFrameSize(from.getMaxFrameSize());
    to.setVirtualHost(from.getVirtualHost());
    to.setSniServerName(from.getSniServerName());
}
 

private void mapNetworkOptions(AmqpProperties from, AmqpClientOptions to) {
    to.setSendBufferSize(from.getSendBufferSize());
    to.setReceiveBufferSize(from.getReceiveBufferSize());
    to.setTrafficClass(from.getTrafficClass());
    to.setReuseAddress(from.isReuseAddress());
    to.setLogActivity(from.isLogActivity());
    to.setReusePort(from.isReusePort());
}
 

@Produces
@Named("my-topic-config")
public AmqpClientOptions options() {
    return new AmqpClientOptions()
            .setHost("localhost")
            .setPort(5672)
            .setUsername("smallrye")
            .setPassword("smallrye");
}
 

@Produces
@Named("my-topic-config2")
public AmqpClientOptions options2() {
    return new AmqpClientOptions()
            .setHost("localhost")
            .setPort(5672)
            .setUsername("smallrye")
            .setPassword("smallrye");
}
 

static AmqpClient createClient(AmqpConnector connector, AmqpConnectorCommonConfiguration config,
        Instance<AmqpClientOptions> instance) {
    AmqpClient client;
    Optional<String> clientOptionsName = config.getClientOptionsName();
    Vertx vertx = connector.getVertx();
    if (clientOptionsName.isPresent()) {
        client = createClientFromClientOptionsBean(vertx, instance, clientOptionsName.get());
    } else {
        client = getClient(vertx, config);
    }
    connector.addClient(client);
    return client;
}
 

static AmqpClient createClientFromClientOptionsBean(Vertx vertx, Instance<AmqpClientOptions> instance,
        String optionsBeanName) {
    Instance<AmqpClientOptions> options = instance.select(NamedLiteral.of(optionsBeanName));
    if (options.isUnsatisfied()) {
        throw ex.illegalStateFindingBean(AmqpClientOptions.class.getName(), optionsBeanName);
    }
    log.createClientFromBean(optionsBeanName);
    return AmqpClient.create(vertx, options.get());
}
 

static AmqpClient getClient(Vertx vertx, AmqpConnectorCommonConfiguration config) {
    try {
        String username = config.getUsername().orElse(null);
        String password = config.getPassword().orElse(null);
        String host = config.getHost();
        int port = config.getPort();
        log.brokerConfigured(host, port, config.getChannel());
        boolean useSsl = config.getUseSsl();
        int reconnectAttempts = config.getReconnectAttempts();
        int reconnectInterval = config.getReconnectInterval();
        int connectTimeout = config.getConnectTimeout();

        // We renamed containerID into container-id. So we must check both.
        String containerId = config.getContainerId()
                .orElseGet(() -> config.config.getOptionalValue("containerId", String.class).orElse(null));

        AmqpClientOptions options = new AmqpClientOptions()
                .setUsername(username)
                .setPassword(password)
                .setHost(host)
                .setPort(port)
                .setContainerId(containerId)
                .setSsl(useSsl)
                .setReconnectAttempts(reconnectAttempts)
                .setReconnectInterval(reconnectInterval)
                .setConnectTimeout(connectTimeout);
        return AmqpClient.create(vertx, options);
    } catch (Exception e) {
        log.unableToCreateClient(e);
        throw ex.illegalStateUnableToCreateClient(e);
    }
}
 

@Produces
@Named("myclientoptions")
public AmqpClientOptions options() {
    return new AmqpClientOptions()
            .setHost(System.getProperty("amqp-host"))
            .setPort(Integer.parseInt(System.getProperty("amqp-port")))
            .setUsername(System.getProperty("amqp-user"))
            .setPassword(System.getProperty("amqp-pwd"));
}
 

private AmqpClientOptions amqpConfig() {
  return new AmqpClientOptions()
    .setHost("localhost")
    .setPort(5672)
    .setUsername("artemis")
    .setPassword("simetraehcapa");
}
 

static AmqpClientOptions amqClientOptions() {
  return new AmqpClientOptions()
    .setHost("localhost")
    .setPort(5672)
    .setUsername("artemis")
    .setPassword("simetraehcapa");
}
 

private void mapNetClientOptions(AmqpProperties from, AmqpClientOptions to) {
    to.setReconnectAttempts(from.getReconnectAttempts());
    to.setReconnectInterval(from.getReconnectInterval());
    to.setHostnameVerificationAlgorithm(from.getHostnameVerificationAlgorithm());
}
 

public AmqpUsage(Vertx vertx, String host, int port, String user, String pwd) {
    this.client = AmqpClient.create(new io.vertx.mutiny.core.Vertx(vertx.getDelegate()),
            new AmqpClientOptions().setHost(host).setPort(port).setUsername(user).setPassword(pwd));
}