import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
private void tryRegisterConsulService(ClusterNodeAddress ownAddress) {

        if (!consul.agentClient().isRegistered(configuration.getConsulServiceId())) {
            log.debug("service {} not registered, will try to do so", configuration.getConsulServiceName());

            Registration.RegCheck ttlCheck = Registration.RegCheck.ttl(configuration.getConsulCheckTTL());
            Registration serviceRegistration = ImmutableRegistration.builder()
                    .name(getServiceName())
                    .address(getNodeAddress(ownAddress))
                    .port(ownAddress.getPort())
                    .id(getServiceId())
                    .addChecks(ttlCheck).build();
            consul.agentClient().register(serviceRegistration, getQueryOptions());

            log.info("registered service {}, registration status now is {}", configuration.getConsulServiceName(), consul.agentClient().isRegistered(configuration.getConsulServiceId()));
        } else {
            log.debug("service {} already registered", configuration.getConsulServiceName());
        }
    }
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
private void registerSelf() throws MalformedURLException {
  if (clientOnly) {
    LOG.debug("Client only. Not registering self with consul");
    return;
  }
  LOG.info("Registering self with Consul");
  Registration.RegCheck ttlCheck = ImmutableRegCheck.builder().ttl("60s").deregisterCriticalServiceAfter("1m").build();
  Registration registration =
    ImmutableRegistration.builder().address(consulServiceAddress).port(Integer.parseInt(consulServicePort))
      .name(consulServiceName).id(consulServiceId).check(ttlCheck).build();
  if (CONSUL_HOLDER != null && CONSUL_HOLDER.hasInstance()) {
    CONSUL_HOLDER.registerSelf(registration);
  } else {
    CONSUL_HOLDER = ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), this);
    CONSUL_HOLDER.registerSelf(registration);
  }
  Runtime.getRuntime().removeShutdownHook(shutdownHook);
  Runtime.getRuntime().addShutdownHook(shutdownHook);
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
synchronized void registerSelf(Registration registration) {
  synchronized (consulLock) {
    if (lastRegisterSelf == null || lastRegisterSelf < (System.currentTimeMillis() - ConsulDiscoveryAgent.pollingInterval)) {
      try {
        consul.agentClient().register(registration);
        lastRegisterSelf = System.currentTimeMillis();
      } catch (ConsulException e) {
        throw new ConsulDiscoveryException(e);
      }
    }
  }
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Override
protected void doPreSetup() throws Exception {
    client = getConsul().agentClient();

    registrations = new ArrayList<>(SERVICE_COUNT);
    expectedBodies = new ArrayList<>(SERVICE_COUNT);

    for (int i = 0; i < SERVICE_COUNT; i++) {
        Registration r = ImmutableRegistration.builder()
            .id("service-" + i)
            .name(SERVICE_NAME)
            .address("127.0.0.1")
            .port(SERVICE_PORT_BASE + i)
            .build();

        client.register(r);

        registrations.add(r);
        expectedBodies.add("ping on " + r.getPort().get());
    }
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Override
public RegCheck buildRegistrationCheck(Map<String, Object> registratorConfig, Address localAddress) {
	
	RegCheck regCheck = null;
	
	try {
		/**
		 * Deal with health check tcp
		 */
		String healthCheckTcp = (String)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_TCP);
		
		if (healthCheckTcp != null && !healthCheckTcp.trim().isEmpty()) {
			
			healthCheckTcp = healthCheckTcp.replaceAll(TCP_TEMPLATE_MYIP, localAddress.getInetAddress().getHostAddress())
					  						 .replaceAll(TCP_TEMPLATE_MYPORT, String.valueOf(localAddress.getPort()));
			
			Long healthCheckTcpIntervalSeconds = Long.valueOf((Integer)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_TCP_INTERVAL_SECONDS));
			regCheck = Registration.RegCheck.tcp(healthCheckTcp, healthCheckTcpIntervalSeconds);
		}
		
	} catch(Exception e) {
		logger.severe("Unexpected error occured trying to build TCP health check : " + e.getMessage(), e);
	}
	
	return regCheck;
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
/**
 * Register the service endpoint with a specific service ID in consul.
 * @param serviceName the name of the service
 * @param serviceId the id of the service
 * @param servicePort the port of the service
 * @param environment the environment of the service (it will always added as a tag in addition)
 * @param healthCheckPath the api path which consul can use for health checks
 * @param healthCheckIntervalSec the interval between consul health checks
 * @param consulCustomTags optionally register more consul tags with the service
 * @throws SecondBaseException if required arguments are missing
 */
public void registerServiceInConsul(
        final String serviceName,
        final String serviceId,
        final int servicePort,
        final String environment,
        final String healthCheckPath,
        final long healthCheckIntervalSec,
        final String... consulCustomTags) throws SecondBaseException {

    if (healthCheckPath == null) {
        throw new SecondBaseException(
                "Must provide health check path to register with consul.");
    }

    // Register service
    final String heathCheckPath = "http://localhost:"
            + servicePort
            + ((healthCheckPath.startsWith("/"))
            ? ""
            : "/")
            + healthCheckPath;
    final Registration.RegCheck serviceRegCheck = Registration.RegCheck.http(
            heathCheckPath,
            healthCheckIntervalSec);
    final String[] tagsArray = Arrays.copyOf(consulCustomTags, consulCustomTags.length + 1);
    tagsArray[tagsArray.length - 1] = environment;

    consulKeepAlive.scheduleAtFixedRate(
            createRegisterTask(
                    servicePort,
                    serviceRegCheck,
                    serviceName,
                    serviceId,
                    tagsArray),
            keepAliveInitialDelaySec,
            keepAlivePeriodSec,
            TimeUnit.SECONDS);
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
private Runnable createRegisterTask(
        final int webconsolePort,
        final Registration.RegCheck regCheck,
        final String serviceName,
        final String serviceId,
        final String... tags) {
    // Attempt to deregister cleanly on shutdown
    deregisterThread.add(serviceId, getConsulClient().agentClient());
    if (!isDeregisterThreadActive.get()) {
        Runtime.getRuntime().addShutdownHook(deregisterThread);
        isDeregisterThreadActive.set(true);
    }
    LOG.info("Registering service in consul. Service name: " + serviceName
            + ". Service ID: "+ serviceId
            + ". Tags: " + Arrays.toString(tags)
            + ". RegCheck: " + regCheck.toString());
    return () -> {
        try {
            if (!getConsulClient().agentClient().isRegistered(serviceId)){
                getConsulClient().agentClient().register(
                        webconsolePort,
                        regCheck,
                        serviceName,
                        serviceId,
                        tags);
            }
        } catch (final Exception e) {
            LOG.warn("Unable contact consul, trying to check " + serviceName, e);
        }
    };
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
private static void register() {
    AgentClient agentClient = consul.agentClient();

    List<Registration.RegCheck> checks = new ArrayList<Registration.RegCheck>();

    HostAndPort serviceHostAndPort = HostAndPort.fromParts(visibleHost, visiblePort);

    Registration.RegCheck mainCheck = Registration.RegCheck.tcp(serviceHostAndPort.toString(), 30);

    checks.add(mainCheck);

    Registration registration = ImmutableRegistration
            .builder()
            .port(visiblePort)
            .address(visibleHost)
            .checks(checks)
            .name(serviceName)
            .id(serviceId)
            .addTags(tsdMode)
            .build();

    agentClient.register(registration);

    if (agentClient.isRegistered(serviceId)) {
        LOGGER.info("Registered this instance with Consul");
    } else {
        LOGGER.warn("Consul reports that this instance is not registered");
    }
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Override
public RegCheck buildRegistrationCheck( Map<String, Object> registratorConfig, Address localAddress) {
	
	RegCheck regCheck = null;
	
	try{
		/**
		 * Deal with health check http
		 */
		String healthCheckHttp = (String)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_HTTP);
		
		if (healthCheckHttp != null && !healthCheckHttp.trim().isEmpty()) {
			
			healthCheckHttp = healthCheckHttp.replaceAll(HTTP_TEMPLATE_MYIP, localAddress.getInetAddress().getHostAddress())
					  						 .replaceAll(HTTP_TEMPLATE_MYPORT, String.valueOf(localAddress.getPort()));
			
			Long healthCheckHttpIntervalSeconds = Long.valueOf((Integer)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_HTTP_INTERVAL_SECONDS));
			regCheck = Registration.RegCheck.http(healthCheckHttp, healthCheckHttpIntervalSeconds);
		}
	
	}catch(Exception e){
		logger.severe("Unexpected error occured trying to build HTTP health check : " + e.getMessage(), e);

	}
	
	return regCheck;
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
private void registerHeartbeat() {
    log.info("registering heartbeat");
    AgentClient agentClient = getConsul().agentClient();
    Registration registration = new Registration();
    registration.setPort(serverProperties.getPort());
    registration.setAddress(dnsResolver.readNonLoopbackLocalAddress());
    registration.setId(toUniqueName("heartbeat"));
    registration.setName(consulProperties.getServiceName());
    registration.setTags(consulProperties.getTags());
    Registration.Check check = new Registration.Check();
    check.setTtl(format("%ss", 2 * (consulProperties.getHeartbeatRate() == null ? DEFAULT_HEARTBEAT_RATE : consulProperties.getHeartbeatRate())));
    registration.setCheck(check);
    agentClient.register(registration);
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Test
public void fullPluginLifecycleWithDefaultConfiguration() throws ExecutionException, InterruptedException {

    // dummy consul client
    Consul consul = mock(Consul.class);
    AgentClient agentClient = mock(AgentClient.class);
    when(consul.agentClient()).thenReturn(agentClient);

    // empty config
    ConfigurationReader configurationReader = mock(ConfigurationReader.class);
    Properties properties = new Properties();
    when(configurationReader.getProperties()).thenReturn(properties);
    Configuration configuration = new Configuration(configurationReader);

    PluginExecutorService pluginExecutorService = mock(PluginExecutorService.class);

    ConsulDiscoveryCallback callback = new ConsulDiscoveryCallback(consul, configuration, pluginExecutorService);
    callback.init("clusternode1", new ClusterNodeAddress("clusternode1-hostname", 1234));

    // check if registration job gets scheduled
    ArgumentCaptor<Runnable> runnableArgument = ArgumentCaptor.forClass(Runnable.class);
    verify(pluginExecutorService, times(2)).scheduleAtFixedRate(runnableArgument.capture(), anyLong(), eq(60l), eq(TimeUnit.SECONDS));

    // run registration job
    Runnable registrationRunnable = runnableArgument.getAllValues().get(0);
    Runnable updateRunnable = runnableArgument.getAllValues().get(1);

    registrationRunnable.run();

    // verify service registration
    ArgumentCaptor<Registration> argument = ArgumentCaptor.forClass(Registration.class);
    verify(agentClient).register(argument.capture(), any());


    Registration registration = argument.getValue();
    assertEquals("cluster-discovery-hivemq", registration.getName());
    assertEquals("clusternode1-hostname", registration.getAddress().get());
    assertEquals(Integer.valueOf(1234), registration.getPort().get());
    assertEquals("cluster-discovery-hivemq", registration.getName());

    String registrationId = registration.getId();
    assertThat(registrationId).containsOnlyDigits();
    assertEquals(1, registration.getChecks().size());

    Registration.RegCheck regCheck = registration.getChecks().get(0);
    assertEquals("120s", regCheck.getTtl().get());

     // run updater job and check consul service pass call
    updateRunnable.run();
    try {
        verify(agentClient).pass(eq(registrationId));
    } catch (NotRegisteredException e) {
        throw new RuntimeException(e);
    }

    HealthClient healthClient = mock(HealthClient.class);
    List<ServiceHealth> nodes = new ArrayList<>();

    Service service = ImmutableService.builder()
            .address("host1")
            .id("not important")
            .service("doesnt matter")
            .port(5678).build();

    Node node = ImmutableNode.builder().node("node1").address("address1").build();
    ServiceHealth serviceHealth = ImmutableServiceHealth.builder().service(service).node(node).build();

    nodes.add(serviceHealth);
    ConsulResponse response = new ConsulResponse(nodes, 0, true, BigInteger.ZERO);
    when(healthClient.getHealthyServiceInstances(anyString(), ArgumentMatchers.<ImmutableQueryOptions>any())).thenReturn(response);
    when(consul.healthClient()).thenReturn(healthClient);

    List<ClusterNodeAddress> addresses = callback.getNodeAddresses().get();
    assertEquals(1, addresses.size());
    assertEquals("host1", addresses.get(0).getHost());
    assertEquals(5678, addresses.get(0).getPort());

    callback.destroy();
    verify(agentClient).deregister(eq(registrationId));
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Override
public void configure(final Env env, final Config config, final Binder binder) throws Throwable {

  Config consulConfig = config.getConfig("consul.default");
  if (!name.equals("default") && config.hasPath("consul." + name)) {
    consulConfig = config.getConfig("consul." + name).withFallback(consulConfig);
  }

  Consul.Builder consulBuilder = Consul.builder()
      .withUrl(consulConfig.getString("url"));

  if (consulBuilderConsumer != null) {
    consulBuilderConsumer.accept(consulBuilder);
  }

  Consul consul = consulBuilder.build();

  env.onStop(consul::destroy);

  env.serviceKey().generate(Consul.class, name, k -> binder.bind(k).toInstance(consul));

  if (consulConfig.hasPath("register")) {

    Config registerConfig = consulConfig.getConfig("register");

    ImmutableRegistration.Builder registrationBuilder = ImmutableRegistration.builder()
        .name(registerConfig.getString("name"))
        .address(registerConfig.getString("host"))
        .port(registerConfig.getInt("port"))
        .tags(registerConfig.getStringList("tags"))
        .id(UUID.randomUUID().toString());

    if (registerConfig.hasPath("check")) {

      Config checkConfig = registerConfig.getConfig("check");

      String http = MessageFormat.format("http://{0}:{1,number,####}{2}",
          registerConfig.getString("host"),
          registerConfig.getInt("port"),
          checkConfig.getString("path"));

      Registration.RegCheck check = Registration.RegCheck.http(http,
          checkConfig.getDuration("interval", TimeUnit.SECONDS),
          checkConfig.getDuration("timeout", TimeUnit.SECONDS));

      registrationBuilder.check(check);

      String response = checkConfig.getString("response");
      env.router().get(checkConfig.getString("path"), () -> response);
    }

    if (registrationBuilderConsumer != null) {
      registrationBuilderConsumer.accept(registrationBuilder);
    }

    Registration registration = registrationBuilder.build();

    AgentClient agentClient = consul.agentClient();
    env.onStarted(() -> agentClient.register(registration));
    env.onStop(() -> agentClient.deregister(registration.getId()));
  }
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Override
public RegCheck buildRegistrationCheck(Map<String, Object> registratorConfig, Address localAddress) {
	
	RegCheck regCheck = null;
	
	try{
		/**
		 * Deal with health check script
		 */
		String rawScript = (String)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_SCRIPT);
		
		if (rawScript != null && !rawScript.trim().isEmpty()) {
			
			Long healthCheckScriptIntervalSeconds = Long.valueOf((Integer)registratorConfig.get(CONFIG_PROP_HEALTH_CHECK_SCRIPT_INTERVAL_SECONDS));
			String healthCheckScript = rawScript.replaceAll(HEALTH_SCRIPT_TEMPLATE_MYIP, localAddress.getInetAddress().getHostAddress())
											    .replaceAll(HEALTH_SCRIPT_TEMPLATE_MYPORT, String.valueOf(localAddress.getPort()));
		
			regCheck = Registration.RegCheck.script(healthCheckScript, healthCheckScriptIntervalSeconds);
		}
	
	}catch(Exception e){
		logger.severe("Unexpected error occured trying to build HTTP health check : " + e.getMessage(), e);
	}
	
	return regCheck;
	
	
	
}
 

import com.orbitz.consul.model.agent.Registration; //导入依赖的package包/类
@Test
public void testServiceDiscovery() throws Exception {
    final AgentClient client = getConsul().agentClient();
    try {
        registrations = new ArrayList<>(3);

        for (int i = 0; i < 3; i++) {
            Registration r = ImmutableRegistration.builder()
                .id("service-" + i)
                .name("my-service")
                .address("127.0.0.1")
                .addTags("a-tag")
                .addTags("key1=value1")
                .addTags("key2=value2")
                .port(9000 + i)
                .build();

            client.register(r);
            registrations.add(r);
        }

        ConsulConfiguration configuration = new ConsulConfiguration();
        configuration.setUrl(consulUrl);
        ServiceDiscovery discovery = new ConsulServiceDiscovery(configuration);

        List<ServiceDefinition> services = discovery.getServices("my-service");
        assertNotNull(services);
        assertEquals(3, services.size());

        for (ServiceDefinition service : services) {
            assertFalse(service.getMetadata().isEmpty());
            assertTrue(service.getMetadata().containsKey("service_name"));
            assertTrue(service.getMetadata().containsKey("service_id"));
            assertTrue(service.getMetadata().containsKey("a-tag"));
            assertTrue(service.getMetadata().containsKey("key1"));
            assertTrue(service.getMetadata().containsKey("key2"));
        }
    } finally {
        if (registrations != null && client != null) {
            registrations.forEach(r -> client.deregister(r.getId()));
        }
    }
}