import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
/**
 * Instantiates a new etcd discoverer and connects to etcd using the given URI.
 * @param uri The URI to connect to etcd.
 */
public EtcdDiscoverer(URI uri) {
    this.myInstances = new HashSet<>();
    this.client = new EtcdClient(uri);

    // Do not retry too many times or wait too long
    this.client.setRetryHandler(new RetryOnce(1));

    // Log server version
    EtcdVersionResponse versionResponse = this.client.version();

    if (versionResponse != null) {
        logger.info("Discoverer connected with etcd at {}, server version {}", uri.toString(), versionResponse.getServer());
    } else {
        logger.warn("Discoverer started, but could not connect to etcd");
    }

    // Create discoverable config directory
    this.client.putDir(buildPath(DISCOVERABLE_CONFIG_DIR));
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
private void processDel(EtcdClient client, String path, boolean dir, Exchange exchange) throws Exception {
    EtcdKeyDeleteRequest request = client.delete(path);
    setRequestTimeout(request, exchange);
    setRequestRecursive(request, exchange);

    if (dir) {
        request.dir();
    }

    try {
        exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
        exchange.getIn().setBody(request.send().get());
    } catch (TimeoutException e) {
        throw new ExchangeTimedOutException(exchange, configuration.getTimeout());
    }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
public EtcdClient createClient() throws Exception {
    String[] uris;
    if (getUris() != null) {
        uris = getUris().split(",");
    } else {
        uris = EtcdConstants.ETCD_DEFAULT_URIS.split(",");
    }

    URI[] etcdUriList = new URI[uris.length];

    int i = 0;
    for (String uri : uris) {
        etcdUriList[i++] = URI.create(camelContext.resolvePropertyPlaceholders(uri));
    }

    return new EtcdClient(
        new EtcdSecurityContext(
            sslContextParameters != null
                ? sslContextParameters.createSSLContext(camelContext)
                : null,
            userName,
            password),
        etcdUriList
    );
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Test
public void testWatchRecovery() throws Exception {
    final String key = "/myKeyRecovery";
    final EtcdClient client = getClient();

    try {
        // Delete the key if present
        client.delete(key).send().get();
    } catch (EtcdException e) {
        if (!e.isErrorCode(EtcdErrorCode.KeyNotFound)) {
            throw e;
        }
    }

    // Fill the vent backlog ( > 1000)
    for (int i = 0; i < 2000; i++) {
        client.put(key, "v" + i).send().get();
    }

    context().startRoute("watchRecovery");

    testWatch("mock:watch-recovery", key, 10);
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
private void testWatch(String mockEndpoint, final String key, int updates) throws Exception {
    final String[] values = new String[updates];
    for (int i = 0; i < updates; i++) {
        values[i] = key + "=myValue-" + i;
    }

    MockEndpoint mock = getMockEndpoint(mockEndpoint);
    mock.expectedMessageCount(2);
    mock.expectedHeaderReceived(EtcdConstants.ETCD_NAMESPACE, EtcdNamespace.watch.name());
    mock.expectedHeaderReceived(EtcdConstants.ETCD_PATH, key);
    mock.expectedBodiesReceived(values);

    final EtcdClient client = getClient();
    for (int i = 0; i < updates; i++) {
        client.put(key, "myValue-" + i).send().get();
    }

    mock.assertIsSatisfied();
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Override
public void deregister() throws Exception {
	EtcdClient etcdClient = null;
	
	try {
		// create our client
		etcdClient = EtcdDiscoveryStrategy.getEtcdClient(etcdUris, etcdUsername, etcdPassword);
		
		// delete
		etcdClient.delete(this.myEtcdKey).send().get();
		
	} catch(Exception e) {
		String msg = "Unexpected error in etcdClient.delete(key:"+this.myEtcdKey+"): " + e.getMessage();
		logger.severe(msg,e);
		throw new Exception(msg,e);
		
	} finally {
		try { etcdClient.close(); } catch(Exception ignore){}
	}
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Test
public void testCustomEtcdNettyClient() throws Exception {
  NioEventLoopGroup evl = new NioEventLoopGroup();
  EtcdNettyConfig config = new EtcdNettyConfig()
      .setConnectTimeout(100)
      .setSocketChannelClass(NioSocketChannel.class)
      .setMaxFrameSize(1024 * 1024)
      .setEventLoopGroup(evl)
      .setHostName("localhost");

  EtcdNettyClient client = new EtcdNettyClient(config, ETCD_URI);
  EtcdClient etcdClient = new EtcdClient(client);
  etcdClient.setRetryHandler(new RetryNTimes(0, 0));

  assertNotNull(etcdClient.version());
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Test
public void testSuccessWithoutRetrying() throws Exception {

    whenHttp(server)
            .match(get("/v2/keys/foo"))
            .then(SUCCESS);

    try (EtcdClient etcd = new EtcdClient(serverURI)) {

        EtcdResponsePromise<EtcdKeysResponse> promise = etcd.get("foo")
                .setRetryPolicy(new RetryNTimes(1, 10))
                .send();

        EtcdKeysResponse resp = promise.get();

        assertThat(resp.node.value).isEqualTo("bar");
        assertThat(promise.getException()).isNull();

    }

    verifyHttp(server).once(
            method(Method.GET),
            uri("/v2/keys/foo")
    );
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Test
public void testSuccessAfterRetrying() throws Exception {

    whenHttp(server)
            .match(get("/v2/keys/foo"))
            .then(sequence(FAILURE, SUCCESS));

    try (EtcdClient etcd = new EtcdClient(serverURI)) {

        EtcdResponsePromise<EtcdKeysResponse> promise = etcd.get("foo")
                .setRetryPolicy(new RetryNTimes(1, 10))
                .send();

        EtcdKeysResponse resp = promise.get();

        assertThat(resp.node.value).isEqualTo("bar");
        assertThat(promise.getException()).isNull();

    }

    verifyHttp(server).times(2,
            method(Method.GET),
            uri("/v2/keys/foo")
    );
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
private static void putFileInEtcd(String path, String content) {
    try(EtcdClient etcd = new EtcdClient(URI.create(getEtcdUrl()))){
        EtcdKeysResponse etcdKeysResponse = etcd.put(directory + path, content).send().get();
        String config = etcdKeysResponse.getNode().getValue();
        assertThat(config).isEqualTo(content);
    } catch (Exception e) {
        throw new RuntimeException("Unknown exception while fetching configuration from etcd", e);
    }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
public EtcdClientWrapper(EtcdClient etcd, String prefix) {
    this.etcd = etcd;
    this.prefix = prefix;
    //possibly we need to create better id ob bus
    this.bus = MessageBusImpl.builder(KvStorageEvent.class, (s) -> new ConditionalMessageBusWrapper<>(s, KvStorageEvent::getKey, KvUtils::predicate))
      .id(getClass().getName())
      .build();
    this.executor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
      .setNameFormat(getClass().getName() + "-bus-%d")
      .setDaemon(true)
      .build());
    eventWhirligig(-1);
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@Bean
public EtcdClientWrapper client() {
    List<URI> uris = new ArrayList<>();
    for (String etcdUrl : etcdUrls) {
        uris.add(URI.create(etcdUrl));
    }
    log.info("About to connect to etcd: {}", (Object)etcdUrls);
    EtcdClient etcd = new EtcdClient(uris.toArray(new URI[uris.size()]));
    return new EtcdClientWrapper(etcd, prefix.trim());
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
/** Init performs the initial read of values from etcd. */
public void init(String flagzDirectory) throws FlagException, EtcdFlagFieldUpdaterException {
  this.directoryPrefix = MoreObjects.firstNonNull(flagzDirectory, directoryFlag.get());
  client = new EtcdClient(uris.toArray(new URI[uris.size()]));
  client.setRetryHandler(retryPolicy);
  initialSetAllFlagz();
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
@BeforeClass
public static void connectEtcd() {
  try {
    client = new EtcdClient(URI.create(ETCD_SERVER));
    client.setRetryHandler(ETCD_RETRY);
    Long etcdIndex = client.getAll().send().get().etcdIndex;
    LOG.info("Connected to Etcd version={} at EtcdIndex({}).", client.version(), etcdIndex);
  } catch (Exception e) {
    throw new RuntimeException("Etc.d must be running on localhost for these tests.", e);
  }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
Object getStats(EtcdClient client) {
    switch (getPath()) {
    case EtcdConstants.ETCD_LEADER_STATS_PATH:
        return client.getLeaderStats();
    case EtcdConstants.ETCD_SELF_STATS_PATH:
        return client.getSelfStats();
    case EtcdConstants.ETCD_STORE_STATS_PATH:
        return client.getStoreStats();
    default:
        throw new IllegalStateException("No stats for " + getPath());
    }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
protected EtcdClient getClient() throws Exception {
    if (client == null) {
        client = ((AbstractEtcdEndpoint)getEndpoint()).createClient();
    }

    return client;
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
private void processSet(EtcdClient client, String path, Exchange exchange) throws Exception {
    EtcdKeyPutRequest request = client.put(path, exchange.getIn().getBody(String.class));
    setRequestTimeToLive(request, exchange);
    setRequestTimeout(request, exchange);

    try {
        exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
        exchange.getIn().setBody(request.send().get());
    } catch (TimeoutException e) {
        throw new ExchangeTimedOutException(exchange, configuration.getTimeout());
    }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
private void processGet(EtcdClient client, String path, Exchange exchange) throws Exception {
    EtcdKeyGetRequest request = client.get(path);
    setRequestTimeout(request, exchange);
    setRequestRecursive(request, exchange);

    try {
        exchange.getIn().setHeader(EtcdConstants.ETCD_NAMESPACE, getNamespace());
        exchange.getIn().setBody(request.send().get());
    } catch (TimeoutException e) {
        throw new ExchangeTimedOutException(exchange, configuration.getTimeout());
    }
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
public EtcdRoutePolicy(EtcdClient client, boolean managedClient) {
    this.client = client;
    this.managedClient = managedClient;
    this.suspendedRoutes =  new HashSet<>();
    this.leader = new AtomicBoolean(false);
    this.lock = new Object();
    this.index = new AtomicLong(0);
    this.serviceName = null;
    this.servicePath = null;
    this.ttl = 60;
    this.watchTimeout = ttl / 3;
    this.shouldStopConsumer = true;
}
 

import mousio.etcd4j.EtcdClient; //导入依赖的package包/类
protected static EtcdClient getEtcdClient(List<URI> etcdUris, String username, String password) throws Exception {
    // build our clients 
    
    if (etcdUris.iterator().next().toString().toLowerCase().indexOf("https") != -1) {
        SslContext sslContext = SslContext.newClientContext();
        return new EtcdClient(sslContext, username, password, etcdUris.toArray(new URI[]{}));
    } else {
        return new EtcdClient(username, password, etcdUris.toArray(new URI[]{}));
    }
}