Hi! This should fix #25.

It follows roughly the approach I mentioned there: a RwLock with a oneshot::channel for telling the spawned tasks to shutdown.

I have some tests on the code I introduced, and the examples all still run. Notably, I've stopped seeing the panics.

Happy for any feedback, thanks!

How do I add a cluster to the client?

let etcd_adress = vec![
        "127.0.0.1:23791".to_owned(),
        "127.0.0.1:23792".to_owned(),
        "127.0.0.1:23793".to_owned(),
    ]
Client::builder().endpoints(etcd_adress.clone()).build()

returns: RpcFailure(RpcStatus { status: Unavailable, details: Some("Name resolution failure") })

but Client::builder().add_endpoint( "127.0.0.1:23791").build() with each of the upper ips works.

FYI this is the docker-compose I use to start the cluster:

version: '3'

services:

  etcd1:
    image: quay.io/coreos/etcd:latest
    environment:
            ETCD_NAME: node1

            ETCD_ADVERTISE_CLIENT_URLS: http://etcd1:2379
            ETCD_LISTEN_CLIENT_URLS: http://0.0.0.0:2379

            ETCD_INITIAL_ADVERTISE_PEER_URLS: http://etcd1:2380
            ETCD_LISTEN_PEER_URLS: http://0.0.0.0:2380

            ETCD_DATA_DIR: /etcd-data/etcd1.etcd
            ETCDCTL_API: 3
            ETCD_DEBUG: 1
            
            ETCD_INITIAL_CLUSTER: node3=http://etcd3:2380,node1=http://etcd1:2380,node2=http://etcd2:2380
            ETCD_INITIAL_CLUSTER_STATE: new
            ETCD_INITIAL_CLUSTER_TOKEN: etcd-ftw           
    ports:
      - 23791:2379
      - 23801:2380


  etcd2:
    image: quay.io/coreos/etcd:latest
    environment:
            ETCD_NAME: node2

            ETCD_INITIAL_ADVERTISE_PEER_URLS: http://etcd2:2380
            ETCD_LISTEN_PEER_URLS: http://0.0.0.0:2380

            ETCD_ADVERTISE_CLIENT_URLS: http://etcd2:2379
            ETCD_LISTEN_CLIENT_URLS: http://0.0.0.0:2379
            
            ETCD_DATA_DIR: /etcd-data/etcd2.etcd
            ETCDCTL_API: 3
            ETCD_DEBUG: 1

            ETCD_INITIAL_CLUSTER: node3=http://etcd3:2380,node1=http://etcd1:2380,node2=http://etcd2:2380
            ETCD_INITIAL_CLUSTER_STATE: new
            ETCD_INITIAL_CLUSTER_TOKEN: etcd-ftw
    ports:
      - 23792:2379
      - 23802:2380


  etcd3:
    image: quay.io/coreos/etcd:latest
    environment:
            ETCD_NAME: node3

            ETCD_INITIAL_ADVERTISE_PEER_URLS: http://etcd3:2380
            ETCD_LISTEN_PEER_URLS: http://0.0.0.0:2380
            
            ETCD_ADVERTISE_CLIENT_URLS: http://etcd3:2379
            ETCD_LISTEN_CLIENT_URLS: http://0.0.0.0:2379
            

            ETCD_DATA_DIR: /etcd-data/etcd3.etcd
            ETCDCTL_API: 3
            ETCD_DEBUG: 1
            
            ETCD_INITIAL_CLUSTER: node3=http://etcd3:2380,node1=http://etcd1:2380,node2=http://etcd2:2380
            ETCD_INITIAL_CLUSTER_STATE: new
            ETCD_INITIAL_CLUSTER_TOKEN: etcd-ftw
    ports:
      - 23793:2379
      - 23803:2380

Thanks in Advance!

at debug model,a lot of debug message print in log:

tonic-0.4.3/src/codec/decode.rs-> line:231]  decoder inner stream error: Status { code: Unknown, message: "h2 protocol error: broken pipe" } 

at the same time, memory using by tonic rise up and stand there,every time interupt appeared,memory usage upgrade.

This PR:

• Upgrades tonic to 0.5
• Upgrades prost to 0.8 (fixes https://rustsec.org/advisories/RUSTSEC-2021-0073.html )
• Bumps version to 0.6.0

Part of the new version of tonic is that they messed up the types for interceptors, so they were moved up an abstraction layer into the Auth/Lease/etc components.

(this adds an export of TxnOp, at least to make the documentation less confusing)

Fixes #43.

Additionally, this removes the take methods on KeyRange and instead makes the fields public. This improves usability, but breaks API compatibility.

Hey :wave:!

Here's what I did in this PR:

• The proto files were updated and a build.rs file to generate Rust code for them was added.
• The requests and responses were updated to support newly added fields.
• A few unwraps were removed to return Result instead.
• Support for using Vec<u8> as keys and values was added.
• Clients are now all Cloneable and requests and responses (and their related types) now are Debugable and Cloneable.
• A KeepAlive stream has been added. It is returned by one of two new methods of LeaseClient: keep_alive and keep_alive_at_interval (keep_alive_once has been removed).

Closes #6

first thanks for your work !

I started etcd ok , the restful can used .

bu when use this client it report:

"grpc-status: Unknown, grpc-message: \"Service was not ready: transport error: buffer\\\'s worker closed unexpectedly\""`', src/etcd_client.rs:179:13
stack backtrace:
   0: backtrace::backtrace::libunwind::trace
             at /Users/runner/.cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.40/src/backtrace/libunwind.rs:88
   1: backtrace::backtrace::trace_unsynchronized
             at /Users/runner/.cargo/registry/src/github.com-1ecc6299db9ec823/backtrace-0.3.40/src/backtrace/mod.rs:66
   2: std::sys_common::backtrace::_print_fmt
             at src/libstd/sys_common/backtrace.rs:77
   3: <std::sys_common::backtrace::_print::DisplayBacktrace as core::fmt::Display>::fmt
             at src/libstd/sys_common/backtrace.rs:59
   4: core::fmt::write
             at src/libcore/fmt/mod.rs:1052
   5: std::io::Write::write_fmt
             at /rustc/41f41b2354778375dc72f7ed1d9323626580dc4d/src/libstd/io/mod.rs:1426
   6: std::io::impls::<impl std::io::Write for alloc::boxed::Box<W>>::write_fmt
             at src/libstd/io/impls.rs:156
   7: std::sys_common::backtrace::_print
             at src/libstd/sys_common/backtrace.rs:62
   8: std::sys_common::backtrace::print
             at src/libstd/sys_common/backtrace.rs:49
   9: std::panicking::default_hook::{{closure}}
             at src/libstd/panicking.rs:204
  10: std::panicking::default_hook
             at src/libstd/panicking.rs:221
  11: std::panicking::rust_panic_with_hook
             at src/libstd/panicking.rs:472
  12: rust_begin_unwind
             at src/libstd/panicking.rs:380
  13: std::panicking::begin_panic_fmt
             at src/libstd/panicking.rs:334
  14: any_client::etcd_client::test_etcd_kv
             at src/etcd_client.rs:179
  15: any_client::etcd_client::test_etcd_kv::{{closure}}
             at src/etcd_client.rs:173
  16: core::ops::function::FnOnce::call_once
             at /rustc/41f41b2354778375dc72f7ed1d9323626580dc4d/src/libcore/ops/function.rs:232
  17: <alloc::boxed::Box<F> as core::ops::function::FnOnce<A>>::call_once
             at /rustc/41f41b2354778375dc72f7ed1d9323626580dc4d/src/liballoc/boxed.rs:1015
  18: __rust_maybe_catch_panic
             at src/libpanic_unwind/lib.rs:86
  19: std::panicking::try
             at /rustc/41f41b2354778375dc72f7ed1d9323626580dc4d/src/libstd/panicking.rs:281
  20: std::panic::catch_unwind
             at /rustc/41f41b2354778375dc72f7ed1d9323626580dc4d/src/libstd/panic.rs:394
  21: test::run_test_in_process
             at src/libtest/lib.rs:539
  22: test::run_test::run_test_inner::{{closure}}
             at src/libtest/lib.rs:452

my code is

// new client 
 pub fn new(addr: &str) -> EtcdCli {
        match Runtime::new()
            .unwrap()
            .block_on(Client::connect(ClientConfig {
                endpoints: vec![addr.to_owned()],
                auth: None,
            })) {
            Ok(c) => EtcdCli { client: c },
            Err(e) => panic!("conn master server has err:{}", e.to_string()),
        }
    }


//set

pub fn set(&self, key: &str, value: &str, ttl: u64) -> ASResult<KValue> {
        let mut req = PutRequest::new(key, value);
        req.set_lease(ttl);
        match Runtime::new().unwrap().block_on(self.client.kv().put(req)) {
            Ok(mut r) => match r.take_prev_kv() {
                Some(kv) => {
                    return Ok(KValue {
                        key: kv.key_str().to_string(),
                        value: kv.value_str().to_string(),
                    })
                }
                None => return Err(err_generic()),
            },
            Err(e) => {
                return Err(err_code(INTERNAL_ERR, e.to_string()));
            }
        }
    }

According to comment of Watcher in etcd repo

In order to prevent a watch stream being stuck in a partitioned node, make sure to wrap context with "WithRequireLeader".

The WithRequireLeader function set the metadata hasleader to true, and etcdctl has done this, so maybe we should do the same? Another way is to provide a function to set metadata of requests, but I believe set hasleader in watch is what etcd recommended since it's crucial to know if the current etcd endpoint is working properly or not.

Hello,

I'm using tonic 0.7.1 for my project but etcd-rs is using tonic 0.6 and this is causing a conflict. Is there a way to address this? It does look like there is some breaking changes, at least in module visibility in tonic.

panic position:

 thread 'main' panicked at 'called `Option::unwrap()` on a `None` value', etcd-rs/src/watch/mod.rs:182:66

watch.tunnel.resp_receiver can not take again:

tunnel.resp_receiver.take()

panicked logs

thread 'tokio-runtime-worker' panicked at 'called `Result::unwrap()` on an `Err` value: Status { code: Unknown, message: "transport error: error trying to connect: tcp connect error: Connection refused (os error 111)" }', .../etcd-rs-0.5.0/src/lease/mod.rs:109:70

code in crate

res = client.lease_keep_alive(request).fuse() => res.unwrap().into_inner()

Looking through the code (v1.0.0-alpha.3), it seems like Client::connect_with_token doesn't actually await anything. Maybe there are plans to do some async work here in the future? If not it would be nice if this could be made sync

We're using this in a scenario where we resolve the endpoints from a dns entry. It would be nice to be able to have something like this baked into the library, so that in the event all nodes become unavailable, the library can recover by querying the dns entry again.