/**
 * Register ports with Consul and retry if unavailable
 *
 * @param applicationScheme Application protocol scheme
 * @param applicationPort Application listening port
 * @param adminPort Administration listening port
 */
void register(String applicationScheme, int applicationPort, int adminPort) {
  try {
    advertiser.register(applicationScheme, applicationPort, adminPort);
    scheduler.ifPresent(ScheduledExecutorService::shutdownNow);
  } catch (ConsulException e) {
    LOGGER.error("Failed to register service in Consul", e);

    retryInterval.ifPresent(
        (interval) -> {
          scheduler.ifPresent(
              (service) -> {
                LOGGER.info(
                    "Will try to register service again in {} seconds", interval.toSeconds());
                service.schedule(
                    () -> register(applicationScheme, applicationPort, adminPort),
                    interval.toSeconds(),
                    TimeUnit.SECONDS);
              });
        });
  }
}
 

@Override
protected Result check() throws Exception {
  try {
    consul.agentClient().ping();
    return Result.healthy();
  } catch (ConsulException e) {
    LOGGER.warn("Unable to ping consul", e);
  }
  return Result.unhealthy("Could not ping consul");
}
 

@Test
public void testCheckUnhealthy() throws Exception {
  doThrow(new ConsulException("error")).when(agent).ping();
  final Result actual = healthCheck.check();
  verify(agent).ping();
  assertThat(actual.isHealthy()).isFalse();
}
 

@Test
public void testDeregisterException() {
  when(agent.isRegistered(anyString())).thenReturn(true);
  doThrow(new ConsulException("error")).when(agent).deregister(anyString());
  advertiser.deregister();
  verify(agent).deregister(anyString());
}
 

@Test
public void testRegister() {
  final ConsulServiceListener listener =
      new ConsulServiceListener(
          advertiser, Optional.of(Duration.milliseconds(1)), Optional.of(scheduler));

  when(advertiser.register(anyString(), anyInt(), anyInt()))
      .thenThrow(new ConsulException("Cannot connect to Consul"))
      .thenReturn(true);

  listener.register(null, 0, 0);

  verify(advertiser, timeout(100).atLeast(1)).register(null, 0, 0);
}
 

@Override
public void prepare() throws ServiceNotProvidedException, ModuleStartException {
    try {
        List<Address> addressList = ConnectUtils.parse(config.getHostPort());

        List<HostAndPort> hostAndPorts = new ArrayList<>();
        for (Address address : addressList) {
            hostAndPorts.add(HostAndPort.fromParts(address.getHost(), address.getPort()));
        }

        Consul.Builder consulBuilder = Consul.builder()
                                             //                    we should set this value or it will be blocked forever
                                             .withConnectTimeoutMillis(3000);

        if (StringUtils.isNotEmpty(config.getAclToken())) {
            consulBuilder.withAclToken(config.getAclToken());
        }

        if (hostAndPorts.size() > 1) {
            client = consulBuilder.withMultipleHostAndPort(hostAndPorts, 5000).build();
        } else {
            client = consulBuilder.withHostAndPort(hostAndPorts.get(0)).build();
        }
    } catch (ConnectStringParseException | ConsulException e) {
        throw new ModuleStartException(e.getMessage(), e);
    }

    ConsulCoordinator coordinator = new ConsulCoordinator(config, client);
    this.registerServiceImplementation(ClusterRegister.class, coordinator);
    this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
}
 

protected void listener() {
    cache = KVCache.newCache(kvClient, clusterId);
    cache.addListener(listener);
    try {
        cache.start();
    } catch (final Throwable e) {
        throw new ConsulException(e.getMessage(), e);
    }
}
 

@Override
public void initialize(Bootstrap<?> bootstrap) {
  // Replace variables with values from Consul KV. Please override
  // getConsulAgentHost() and getConsulAgentPort() if Consul is not
  // listening on the default localhost:8500.
  try {
    LOGGER.debug("Connecting to Consul at {}:{}", getConsulAgentHost(), getConsulAgentPort());

    final Consul.Builder builder =
        Consul.builder()
            .withHostAndPort(HostAndPort.fromParts(getConsulAgentHost(), getConsulAgentPort()));

    getConsulAclToken()
        .ifPresent(
            token -> {
              // setting both ACL token here and with header, supplying an
              // auth header. This should cover both use cases: endpoint
              // supports legacy ?token query param and other case
              // in which endpoint requires an X-Consul-Token header.
              // @see https://www.consul.io/api/index.html#acls

              LOGGER.debug("Using Consul ACL token: {}", token);

              builder
                  .withAclToken(token)
                  .withHeaders(ImmutableMap.of(CONSUL_AUTH_HEADER_KEY, token));
            });

    // using Consul as a configuration substitution provider
    bootstrap.setConfigurationSourceProvider(
        new SubstitutingSourceProvider(
            bootstrap.getConfigurationSourceProvider(),
            new ConsulSubstitutor(builder.build(), strict, substitutionInVariables)));

  } catch (ConsulException e) {
    LOGGER.warn(
        "Unable to query Consul running on {}:{}," + " disabling configuration substitution",
        getConsulAgentHost(),
        getConsulAgentPort(),
        e);
  }
}