org.springframework.boot.test.context.runner.WebApplicationContextRunner#com.ecwid.consul.v1.ConsulClient源码实例Demo

@BeforeClass
public static void beforeClass() {

	assumeTrue(CanConnect.to(new InetSocketAddress(CONSUL_HOST, CONSUL_PORT)));

	ConsulClient client = new ConsulClient();

	Response<List<CatalogService>> response = client.getCatalogService("vault",
			QueryParams.DEFAULT);

	if (response.getValue().isEmpty()) {

		NewService service = new NewService();
		service.setAddress("localhost");
		service.setPort(8200);
		service.setId("vault");
		service.setName("vault");

		client.agentServiceRegister(service);
	}
}
 

public void setAvailableConsulClient() {
    if (consulClient != null) {
        return;
    }
    for (Map.Entry<String, ConsulClient> entry : consulClientMap.entrySet()) {
        try {
            if (entry.getValue().getStatusLeader().getValue() != null) {
                String[] hostPort = entry.getKey().split(":");
                consulProperties.setHost(hostPort[0]);
                consulProperties.setPort(Integer.parseInt(hostPort[1]));
                consulClient = entry.getValue();
                break;
            }
        } catch (Exception e) {
            log.error("Occur an error when executing method 'setAvailableConsulClient'.", e);
            continue;
        }
    }
}
 

@Test
public void firstCallDoesNotPublishEvent() {
	ApplicationEventPublisher eventPublisher = mock(ApplicationEventPublisher.class);
	this.configProperties.setFormat(FILES);

	GetValue getValue = new GetValue();
	String context = "/config/app.yml";
	ConsulClient consul = mock(ConsulClient.class);
	List<GetValue> getValues = Collections.singletonList(getValue);

	Response<List<GetValue>> response = new Response<>(getValues, 1L, false, 1L);
	when(consul.getKVValues(eq(context), anyString(), any(QueryParams.class)))
			.thenReturn(response);

	ConfigWatch watch = new ConfigWatch(this.configProperties, consul,
			new LinkedHashMap<String, Long>());
	watch.setApplicationEventPublisher(eventPublisher);

	watch.watchConfigKeyValues();
	verify(eventPublisher, times(0)).publishEvent(any(RefreshEvent.class));
}
 

@Test
public void testStop() {
    LinkedHashMap<String, Long> initialIndexes = new LinkedHashMap<>();
    initialIndexes.put("/context", 0L);
    ConfigWatch watch = new ConfigWatch(this.configProperties, new ConsulClient(), bmsAuthClient, initialIndexes);
    watch.start();
    Assert.assertTrue(watch.isRunning());
    Runnable runnable = new Runnable() {
        @Override
        public void run() {
            // do nothing
        }
    };
    watch.stop(runnable);
    Assert.assertFalse(watch.isRunning());
}
 

@Test
public void firstCallDoesNotPublishEvent() {
    ApplicationEventPublisher eventPublisher = mock(ApplicationEventPublisher.class);
    this.configProperties.setFormat(FILES);

    GetValue getValue = new GetValue();
    String context = "/config/app.yml";
    ConsulClient consul = mock(ConsulClient.class);
    List<GetValue> getValues = Collections.singletonList(getValue);

    Response<List<GetValue>> response = new Response<>(getValues, 1L, false, 1L);
    when(consul.getKVValues(ArgumentMatchers.eq(context), ArgumentMatchers.anyString(),
            ArgumentMatchers.any(QueryParams.class)))
            .thenReturn(response);

    ConfigWatch watch = new ConfigWatch(this.configProperties, consul, bmsAuthClient,
            new LinkedHashMap<String, Long>());
    watch.setApplicationEventPublisher(eventPublisher);

    watch.watchConfigKeyValues(context);
    verify(eventPublisher, times(0)).publishEvent(ArgumentMatchers.any(RefreshEvent.class));
}
 

@Before
public void setup() {
    this.properties = new ConsulProperties();
    this.client = new ConsulClient(this.properties.getHost(),
            this.properties.getPort());
    this.client.deleteKVValues(PREFIX);
    this.client.setKVValue(KEY1, VALUE1);
    this.client.setKVValue(KEY2, VALUE2);

    this.context = new SpringApplicationBuilder(Config.class)
            .web(WebApplicationType.NONE).run("--SPRING_APPLICATION_NAME=" + APP_NAME,
                    "--spring.cloud.consul.config.prefix=" + PREFIX,
                    "--spring.cloud.consul.config.token-enabled=false",
                    "--spring.cloud.consul.config.system-labels=app-test,env-test,d-test",
                    "spring.cloud.consul.config.watch.delay=10");

    this.client = this.context.getBean(ConsulClient.class);
    this.properties = this.context.getBean(ConsulProperties.class);
    this.environment = this.context.getEnvironment();
}
 

@Before
public void setup() {
	this.properties = new ConsulProperties();
	this.client = new ConsulClient(this.properties.getHost(),
			this.properties.getPort());
	this.client.setKVValue(ROOT + APPLICATION_YML, "foo: bar\nmy.baz: ${foo}");
	this.client.setKVValue(ROOT + APPLICATION_DEV_YML,
			"foo: bar-dev\nmy.baz: ${foo}");
	this.client.setKVValue(ROOT + "/master.ref", UUID.randomUUID().toString());
	this.client.setKVValue(ROOT + APP_NAME_PROPS, "foo: bar-app\nmy.baz: ${foo}");
	this.client.setKVValue(ROOT + APP_NAME_DEV_PROPS,
			"foo: bar-app-dev\nmy.baz: ${foo}");

	this.context = new SpringApplicationBuilder(Config.class)
			.web(WebApplicationType.NONE).run("--spring.application.name=" + APP_NAME,
					"--spring.cloud.consul.config.prefix=" + ROOT,
					"--spring.cloud.consul.config.format=FILES",
					"--spring.profiles.active=dev",
					"spring.cloud.consul.config.watch.delay=1");

	this.client = this.context.getBean(ConsulClient.class);
	this.properties = this.context.getBean(ConsulProperties.class);
	this.environment = this.context.getEnvironment();
}
 

@Before
public void init() {
    this.consul = ConsulStarterBuilder.consulStarter()
        .build()
        .start();
    int port = consul.getHttpPort();
    String host = "127.0.0.1";
    client = new ConsulClient(host, port);
    Converter<String, List<FlowRule>> flowConfigParser = buildFlowConfigParser();
    String flowRulesJson =
        "[{\"resource\":\"test\", \"limitApp\":\"default\", \"grade\":1, \"count\":\"0.0\", \"strategy\":0, "
            + "\"refResource\":null, "
            +
            "\"controlBehavior\":0, \"warmUpPeriodSec\":10, \"maxQueueingTimeMs\":500, \"controller\":null}]";
    initConsulRuleData(flowRulesJson);
    rules = flowConfigParser.convert(flowRulesJson);
    consulDataSource = new ConsulDataSource<>(host, port, ruleKey, waitTimeoutInSecond, flowConfigParser);
    FlowRuleManager.register2Property(consulDataSource.getProperty());
}
 

@PostConstruct
public void init() {
    consulClient = new ConsulClient(agentHost);
    executor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            try {
                log.info("begin to load from registry");
                servicesPassing.clear();
                servicesFailing.clear();
                loadAllServiceFromConsul();
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            }
        }
    }, 0, 1, TimeUnit.MINUTES);
}
 

@Before
public void setup() {
	this.properties = new ConsulProperties();
	this.client = new ConsulClient(this.properties.getHost(),
			this.properties.getPort());
	this.client.deleteKVValues(PREFIX);
	this.client.setKVValue(KEY1, VALUE1);
	this.client.setKVValue(KEY2, VALUE2);

	this.context = new SpringApplicationBuilder(Config.class)
			.web(WebApplicationType.NONE)
			.run("--spring.application.name=testConsulPropertySourceLocatorAppNameCustomized",
					"--spring.cloud.consul.config.name=" + CONFIG_NAME,
					"--spring.cloud.consul.config.prefix=" + ROOT);

	this.client = this.context.getBean(ConsulClient.class);
	this.properties = this.context.getBean(ConsulProperties.class);
	this.environment = this.context.getEnvironment();
}
 

@Override
public void run() {
    Lock lock = new Lock(new ConsulClient(), "lock-key", checkTtl);
    try {
        // 获取分布式互斥锁（参数含义：阻塞模式、每次尝试获取锁的间隔500ms、尝试n次）
        if (lock.lock(true, 500L, null)) {
            log.info("Thread {} start!", flag);
            // 处理业务逻辑
            Thread.sleep(new Random().nextInt(5000));
            log.info("Thread {} end!", flag);
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        lock.unlock();
    }

}
 

@Before
public void setup() {
	this.properties = new ConsulProperties();
	this.client = new ConsulClient(this.properties.getHost(),
			this.properties.getPort());
	this.client.deleteKVValues(PREFIX);
	this.client.setKVValue(KEY1, VALUE1);
	this.client.setKVValue(KEY2, VALUE2);

	this.context = new SpringApplicationBuilder(Config.class)
			.web(WebApplicationType.NONE).run("--SPRING_APPLICATION_NAME=" + APP_NAME,
					"--spring.cloud.consul.config.prefix=" + ROOT,
					"spring.cloud.consul.config.watch.delay=10");

	this.client = this.context.getBean(ConsulClient.class);
	this.properties = this.context.getBean(ConsulProperties.class);
	this.environment = this.context.getEnvironment();
}
 

private void testAutoRegistrationDisabled(String testName, String disableProperty) {
	new WebApplicationContextRunner().withUserConfiguration(TestConfig.class)
			.withPropertyValues("spring.application.name=" + testName,
					disableProperty + "=false", "server.port=0")
			.run(context -> {
				assertThat(context).doesNotHaveBean(ConsulServiceRegistry.class);
				assertThat(context).doesNotHaveBean(HeartbeatProperties.class);

				ConsulClient consul = context.getBean(ConsulClient.class);

				Response<Map<String, Service>> response = consul.getAgentServices();
				Map<String, Service> services = response.getValue();
				Service service = services.get(testName);
				assertThat(service).as("service was registered").isNull();
			});
}
 

@Before
public void setup() {
	this.properties = new ConsulProperties();
	this.prefix = "consulPropertySourceTests"
			+ new Random().nextInt(Integer.MAX_VALUE);
	this.client = new ConsulClient(this.properties.getHost(),
			this.properties.getPort());
}
 

private void initConsulCluster() {
    try {
        if (consulServers != null && consulServers.length() > 0) {
            for (String hostPorts : consulServers.split(",")) {
                String[] hostPort = hostPorts.trim().split(":");
                consulClientMap.put(hostPorts, new ConsulClient(hostPort[0], Integer.parseInt(hostPort[1])));
            }
        }
    } catch (Exception e) {
        // It's not a strong dependency to Consul.
        log.error("Could not connect to Consul, servers = [{}].", consulServers, e);
    }
}
 

public void deleteAlert(Cluster cluster, PrometheusAlert alert) {
    ClusterManager ambari = cluster.getClusterManager();
    try {
        TlsConfiguration tlsConfig = tlsSecurityService.getTls(cluster.getId());
        ConsulClient consulClient = ConsulUtils.createClient(ambari.getHost(), cluster.getPort(), tlsConfig);
        String alertKey = getKeyNameForAlert(alert);
        consulClient.deleteKVValue(alertKey);
        LOGGER.debug("Alert has been removed from Consul KV store with name: '{}' on host: '{}'.", alertKey, ambari.getHost());
    } catch (Exception e) {
        LOGGER.error("Alert could not be deleted from Consul KV store: {}", e.getMessage());
    }
}
 

public ConfigWatch(ConsulConfigProperties properties, ConsulClient consul, BmsAuthClient bmsAuthClient,
                   LinkedHashMap<String, Long> initialIndexes, List<ThreadPoolTaskScheduler> taskSchedulers) {
    this.properties = properties;
    this.consul = consul;
    this.consulIndexes = new LinkedHashMap<>(initialIndexes);
    this.watchFutures = new ScheduledFuture<?>[consulIndexes.size()];
    this.bmsAuthClient = bmsAuthClient;
    if (taskSchedulers == null) {
        this.taskSchedulers = getTaskSchedulers();
    } else {
        this.taskSchedulers = taskSchedulers;
    }
}
 

public ConsulPropertySourceLocator(ConsulClient consul,
                                   BmsAuthClient bmsAuthClient,
                                   ConsulConfigProperties properties) {
    this.consul = consul;
    this.bmsAuthClient = bmsAuthClient;
    this.properties = properties;
}
 

@Bean
@ConditionalOnProperty(name = "spring.cloud.consul.config.watch.enabled",
        matchIfMissing = true)
public ConfigWatch configWatch(ConsulConfigProperties properties, BmsAuthClient bmsAuthClient,
                               ConsulPropertySourceLocator locator, ConsulClient consul) {
    return new ConfigWatch(properties, consul, bmsAuthClient, locator.getContextIndexes());
}
 

@Bean
@Primary
public ConsulClient consulClient(ConsulProperties consulProperties,
                                 ConsulConfigProperties consulConfigProperties) {
    final int agentPort = consulProperties.getPort();
    final String agentHost = !StringUtils.isEmpty(consulProperties.getScheme())
            ? consulProperties.getScheme() + "://" + consulProperties.getHost()
            : consulProperties.getHost();

    logger.info("Init consul host: " + agentHost + " port: " + agentPort);
    if (consulProperties.getTls() != null) {
        ConsulProperties.TLSConfig tls = consulProperties.getTls();
        TLSConfig tlsConfig = new TLSConfig(
                tls.getKeyStoreInstanceType(),
                tls.getCertificatePath(),
                tls.getCertificatePassword(),
                tls.getKeyStorePath(),
                tls.getKeyStorePassword()
        );
        return new ConsulClient(agentHost, agentPort, tlsConfig);
    }
    HttpRequestInterceptor httpRequestInterceptor = new BmsCommonInterceptor();
    BmsHttpTransport httpTransport = new BmsHttpTransport(httpRequestInterceptor);
    ConsulRawClient rawClient = new ConsulRawClient(httpTransport.getHttpClient(),
            agentHost, agentPort, consulConfigProperties.getPath());
    return new ConsulClient(rawClient);
}
 

public ConsulPropertySource(String context, ConsulClient source, BmsAuthClient bmsAuthClient,
                            ConsulConfigProperties configProperties) {
    super(context, source);
    this.context = context;
    this.configProperties = configProperties;
    this.bmsAuthClient = bmsAuthClient;
}
 

@Before
public void setup() {


    this.properties = new ConsulProperties();
    this.client = new ConsulClient(this.properties.getHost(),
            this.properties.getPort());
    this.client.setKVValue(ROOT + APPLICATION_YML, "foo: bar\nmy.baz: ${foo}");
    this.client.setKVValue(ROOT + APPLICATION_DEV_YML,
            "foo: bar-dev\nmy.baz: ${foo}");
    this.client.setKVValue(ROOT + "/master.ref", UUID.randomUUID().toString());
    this.client.setKVValue(ROOT + APP_NAME_PROPS, "foo: bar-app\nmy.baz: ${foo}");
    this.client.setKVValue(ROOT + APP_NAME_DEV_PROPS,
            "foo: bar-app-dev\nmy.baz: ${foo}");

    this.context = new SpringApplicationBuilder(ConsulConfigBootstrapConfiguration.class)
            .web(WebApplicationType.NONE).run("--spring.application.name=" + APP_NAME,
                    "--spring.cloud.consul.config.prefix=" + ROOT,
                    "--spring.cloud.consul.config.format=FILES",
                    "--spring.cloud.consul.config.token-enabled=false",
                    "--spring.cloud.consul.config.default-context=application",
                    "--spring.cloud.consul.config.system-labels=application-dev,testFilesFormat," +
                            "testFilesFormat-dev",
                    "spring.cloud.consul.config.watch.delay=1");

    this.client = this.context.getBean(ConsulClient.class);
    this.properties = this.context.getBean(ConsulProperties.class);
    this.environment = this.context.getEnvironment();
}
 

private void setupWatch(ApplicationEventPublisher eventPublisher, String context,
                        Response<List<GetValue>> response) {
    ConsulClient consul = mock(ConsulClient.class);

    when(consul.getKVValues(ArgumentMatchers.eq(context), nullable(String.class),
            ArgumentMatchers.any(QueryParams.class))).thenReturn(response);

    LinkedHashMap<String, Long> initialIndexes = new LinkedHashMap<>();
    initialIndexes.put(context, 0L);
    startWatch(eventPublisher, consul, initialIndexes);
}
 

private void setupWatchThrowException(ApplicationEventPublisher eventPublisher, String context) {
    ConsulClient consul = mock(ConsulClient.class);
    OperationException operationException = new OperationException(403, null, null);
    when(consul.getKVValues(ArgumentMatchers.eq(context), nullable(String.class),
            ArgumentMatchers.any(QueryParams.class))).thenThrow(operationException);

    LinkedHashMap<String, Long> initialIndexes = new LinkedHashMap<>();
    initialIndexes.put(context, 0L);
    startWatch(eventPublisher, consul, initialIndexes);
}
 

private void setupWatchWithIllegalIndex(ApplicationEventPublisher eventPublisher, String context,
                                        Response<List<GetValue>> response) {
    ConsulClient consul = mock(ConsulClient.class);

    when(consul.getKVValues(ArgumentMatchers.eq(context), nullable(String.class),
            ArgumentMatchers.any(QueryParams.class))).thenReturn(response);

    LinkedHashMap<String, Long> initialIndexes = new LinkedHashMap<>();
    initialIndexes.put(context, -1L);
    startWatch(eventPublisher, consul, initialIndexes);
}
 

@Test
public void testGetMethods() {
    LinkedHashMap<String, Long> initialIndexes = new LinkedHashMap<>();
    initialIndexes.put("/context", 0L);
    ConfigWatch watch = new ConfigWatch(this.configProperties, new ConsulClient(), bmsAuthClient, initialIndexes);
    Assert.assertEquals(0, watch.getPhase());
    Assert.assertEquals(true, watch.isAutoStartup());
}
 

/**
 * Fetches healthy garmadon forwarder end points from consul.
 */
private List<HealthService> getHealthyEndPoints() {
    ConsulClient client = new ConsulClient("localhost");

    HealthServicesRequest request = HealthServicesRequest.newBuilder()
            .setPassing(true)
            .setQueryParams(QueryParams.DEFAULT)
            .build();
    Response<List<HealthService>> healthyServices = client.getHealthServices(serviceName, request);

    return healthyServices.getValue();
}
 

/**
 * Fetches healthy service nodes
 */
private List<HealthService> getHealthyEndPoints(String serviceName) {
    ConsulClient client = new ConsulClient("localhost");

    HealthServicesRequest request = HealthServicesRequest.newBuilder()
        .setPassing(true)
        .setQueryParams(QueryParams.DEFAULT)
        .build();
    Response<List<HealthService>> healthyServices = client.getHealthServices(serviceName, request);

    return healthyServices.getValue();
}
 

@Override
ConsulClient createServer(String clusterId, Supplier<String> serverAddressSupplier, String namespace) throws Exception {
    String serverAddress = serverAddressSupplier.get();
    serverAddress = serverAddress.startsWith(HTTP) ? serverAddress : HTTP + serverAddress;
    URL url = new URL(serverAddress);
    return new ConsulClient(url.getHost(), url.getPort());
}
 

@Override
public boolean delete(TaskDO taskDO) {
    try {

        NamingService sourceNamingService =
            nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getGroupName());
        ConsulClient consulClient = consulServerHolder.get(taskDO.getDestClusterId(), taskDO.getGroupName());

        sourceNamingService.unsubscribe(taskDO.getServiceName(), nacosListenerMap.get(taskDO.getTaskId()));

        // 删除目标集群中同步的实例列表
        Response<List<HealthService>> serviceResponse =
            consulClient.getHealthServices(taskDO.getServiceName(), true, QueryParams.DEFAULT);
        List<HealthService> healthServices = serviceResponse.getValue();
        for (HealthService healthService : healthServices) {

            if (needDelete(ConsulUtils.transferMetadata(healthService.getService().getTags()), taskDO)) {
                consulClient.agentServiceDeregister(healthService.getService().getId());
            }
        }
    } catch (Exception e) {
        log.error("delete a task from nacos to nacos was failed, taskId:{}", taskDO.getTaskId(), e);
        metricsManager.recordError(MetricsStatisticsType.DELETE_ERROR);
        return false;
    }
    return true;
}