import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
List<ServiceHealth> getServices(String consulServiceName) {
  synchronized (consulLock) {
    checkInstance();
    try {
      ServiceCache cached = cachedServices.get(consulServiceName);
      if (cached == null || cached.when < (System.currentTimeMillis() - ConsulDiscoveryAgent.pollingInterval)) {
        HealthClient healthClient = consul.healthClient();
        List<ServiceHealth> services = healthClient.getAllServiceInstances(consulServiceName).getResponse();
        cached = new ServiceCache(System.currentTimeMillis(), services);
        cachedServices.put(consulServiceName, cached);
      }
      return cached.services;
    } catch (ConsulException e) {
      throw new ConsulDiscoveryException(e);
    }
  }
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public ListenableFuture<List<ClusterNodeAddress>> getNodeAddresses() {

    final List<ClusterNodeAddress> addresses = new ArrayList<>();

    List<ServiceHealth> nodes = consul.healthClient()
            .getHealthyServiceInstances(configuration.getConsulServiceName(), getQueryOptions())
            .getResponse();

    for (ServiceHealth node : nodes) {
        addresses.add(new ClusterNodeAddress(node.getService().getAddress(), node.getService().getPort()));
    }

    log.debug(String.format("consul returned %s cluster nodes (%s)",
            addresses.size(),
            addresses.stream().map(a -> a.getHost() + ":" + a.getPort())
                    .collect(Collectors.joining(", "))));
    return Futures.immediateFuture(addresses);
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public void notify(Map<HostAndPort, ServiceHealth> newValues) {
    Set<Registration> previousEntries = topologyManager.registrationsForService(this.name);

    Set<Registration> newEntries = newValues.values().stream()
            .map(e -> new Registration("consul",
                                       this.name,
                                       e.getService().getAddress(),
                                       e.getService().getPort())
                    .addTags(e.getService().getTags())
            )
            .collect(Collectors.toSet());

    previousEntries.stream()
            .filter(e -> !newEntries.contains(e))
            .forEach(e -> {
                this.topologyManager.unregister(e);
            });

    newEntries.stream()
            .filter(e -> !previousEntries.contains(e))
            .forEach(e -> {
                this.topologyManager.register(e);
            });
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
private void connect()
{
    if (connected) {
        return;
    }
    connected = true;
    // get all healthy docker services
    List<ServiceHealth> services = consulClient
        .healthClient()
        .getHealthyServiceInstances(serviceName)
        .getResponse();
    for (ServiceHealth service : services) {
        dockerMetadataCollectors.put(
            service.getNode().getNode(),
            dockerMetadataCollectorProvider.getCollector(
                URI.create("docker://" + service.getNode().getAddress() + ":" + service.getService().getPort())
            )
        );
    }
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public void notify(Map<HostAndPort, ServiceHealth> newValues) {
    Set<Registration> previousEntries = topologyManager.registrationsForService(this.name);

    Set<Registration> newEntries = newValues.values().stream()
            .map(e -> new Registration("consul",
                    this.name,
                    e.getService().getAddress(),
                    e.getService().getPort())
                    .addTags(e.getService().getTags())
            )
            .collect(Collectors.toSet());

    previousEntries.stream()
            .filter(e -> !newEntries.contains(e))
            .forEach(e -> {
                this.topologyManager.unregister(e);
            });

    newEntries.stream()
            .filter(e -> !previousEntries.contains(e))
            .forEach(e -> {
                this.topologyManager.register(e);
            });
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public void start() throws Exception {
  LOG.info("Starting Consul Discovery Agent for " + basePath);
  taskRunner = new TaskRunnerFactory("Consul Discovery Agent");
  taskRunner.init();
  running.set(true);

  if (CONSUL_HOLDER == null || !CONSUL_HOLDER.hasInstance()) {
    CONSUL_HOLDER = ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), this);
  } else {
    CONSUL_HOLDER = ConsulHolder.instance();
  }
  registerSelf();

  executorService = Executors.newSingleThreadScheduledExecutor();
  executorService.scheduleWithFixedDelay(new Runnable() {
    @Override
    public void run() {
      if (running.get()) {
        try {
          if (CONSUL_HOLDER == null || !CONSUL_HOLDER.hasInstance()) {
            CONSUL_HOLDER =
              ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), ConsulDiscoveryAgent.this);
          }
          checkTtl();

          List<ServiceHealth> services = CONSUL_HOLDER.getServices(consulServiceName);
          handleServices(services);
        } catch (Throwable t) {
          //We don't really want this thread to ever die as long as the agent is running, so we'll just log any
          // errors.
          LOG.warn("Failure in Consul service discovery", t);
        }
      } else {
        Thread.currentThread().interrupt();
      }
    }
  }, 0, pollingInterval, TimeUnit.MILLISECONDS);
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@GET
@Timed
@Path("/consul/{service}")
public List<ServiceHealth> getHealthyServiceInstances(
        @PathParam("service") String service) {
    return consul.healthClient().getHealthyServiceInstances(service)
            .getResponse();
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public List<ServiceCallServer> getUpdatedListOfServers(String name) {
    List<CatalogService> services = getCatalogClient()
        .getService(name, getCatalogOptions())
        .getResponse();

    List<ServiceHealth> healths = getHealthClient()
        .getAllServiceInstances(name, getCatalogOptions())
        .getResponse();

    return services.stream()
        .filter(service -> !hasFailingChecks(service, healths))
        .map(this::newServer)
        .collect(Collectors.toList());
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
ServiceCache(long when, List<ServiceHealth> services) {
  this.when = when;
  this.services = services;
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public Collection<ServiceHealth> discover(@Nonnull final Consul consul) {
    return consul.healthClient().getHealthyServiceInstances(service)
            .getResponse();
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
protected boolean isNotHealthy(ServiceHealth health) {
    return health.getChecks().stream().anyMatch(this::isNotHealthy);
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
protected boolean isCheckOnService(ServiceHealth check, CatalogService service) {
    return check.getService().getService().equalsIgnoreCase(service.getServiceName());
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
protected boolean hasFailingChecks(CatalogService service, List<ServiceHealth> healths) {
    return healths.stream().anyMatch(health -> isCheckOnService(health, service) && isNotHealthy(health));
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Override
public Iterable<DiscoveryNode> discoverNodes() {
	
	List<DiscoveryNode> toReturn = new ArrayList<DiscoveryNode>();
	
	try {
		// discover healthy nodes only? (and its NOT the first invocation...)
		if (this.consulHealthyOnly && discoverNodesInvoked) {
			
			List<ServiceHealth> nodes = consulHealthClient.getHealthyServiceInstances(consulServiceName, ConsulUtility.getAclToken(this.consulAclToken)).getResponse();
			
			for (ServiceHealth node : nodes) {
				toReturn.add(new SimpleDiscoveryNode(
								new Address(node.getService().getAddress(),node.getService().getPort())));
				getLogger().info("Discovered healthy node: " + node.getService().getAddress()+":"+node.getService().getPort());
			}
			
		// discover all services, regardless of health or this is the first invocation
		} else {
			
			ConsulResponse<List<CatalogService>> response = this.consulCatalogClient.getService(consulServiceName, ConsulUtility.getAclToken(this.consulAclToken));
			
			for (CatalogService service : response.getResponse()) {
				
				String discoveredAddress = null;
				String rawServiceAddress = service.getServiceAddress();
				String rawAddress = service.getAddress();
				
				if (rawServiceAddress != null && !rawServiceAddress.trim().isEmpty()) {
				    discoveredAddress = rawServiceAddress;
					
				} else if (rawAddress != null && !rawAddress.trim().isEmpty()) {
				    getLogger().warning("discoverNodes() ServiceAddress was null/blank! " +
						     "for service: " + service.getServiceName() + 
						     " falling back to Address value");
				    discoveredAddress = rawAddress;
					
				} else {
				    getLogger().warning("discoverNodes() could not discover an address, " +
						     "both ServiceAddress and Address were null/blank! " +
						     "for service: " + service.getServiceName());
				}
				
				toReturn.add(new SimpleDiscoveryNode(
						new Address(discoveredAddress, service.getServicePort())));
				getLogger().info("Discovered healthy node: " + discoveredAddress+":"+service.getServicePort());
			}
		}
		
	} catch(Exception e) {
		getLogger().severe("discoverNodes() unexpected error: " + e.getMessage(),e);
	}

	// flag we were invoked
	discoverNodesInvoked = true;
	
	return toReturn;
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
@Test
public void findAvailableServicesTest() {
    injects();
    final List<ServiceHealth> nodes = healthClient.getHealthyServiceInstances("DataService").getResponse();
    Assert.assertTrue(CollectionUtils.isEmpty(nodes));
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
/**
 * Converts a list of {@link ServiceHealth} objects into {@link Server}
 * objects
 *
 * @param services
 *            list of healthy service instances
 * @return list of server instances
 */
private List<Server> buildServerList(
        @Nonnull final Collection<ServiceHealth> services) {
    return services.stream().map(this::buildServer)
            .collect(Collectors.toList());
}
 

import com.orbitz.consul.model.health.ServiceHealth; //导入依赖的package包/类
Collection<ServiceHealth> discover(Consul consul);