// The etcd v3 KV/Lease/Watch gRPC service calls, layered on the etcdserverpb messages
// and moonzero's `RpcChannel` (the h2c gRPC client that carries the real HTTP/2 frames
// through the transport). This is the network client go-zero's `discov` runs against a
// live etcd: `range` to list a prefix, `put` under a lease to register, `delete_range`
// to deregister, `lease_grant` for the instance TTL, and `watch` to stream changes.
///|
/// An etcd gRPC call that returned a non-OK `grpc-status`.
pub suberror EtcdError {
EtcdError(String)
}
///|
/// An etcd v3 client bound to a gRPC channel (real etcd, or a mock server in tests).
pub struct EtcdClient {
channel : RpcChannel
}
///|
/// An etcd client over `channel`.
pub fn EtcdClient::new(channel : RpcChannel) -> EtcdClient {
{ channel, }
}
///|
/// Turn a unary reply into decoded bytes, mapping a non-OK status to `EtcdError`.
fn etcd_unwrap(
result : Result[Bytes, @moonrpc.Status],
call_name : String,
) -> Bytes raise EtcdError {
match result {
Ok(reply) => reply
Err(status) =>
raise EtcdError(call_name + " failed: " + @moonrpc.Status::name(status))
}
}
///|
/// `KV.Range`: read the key or prefix range in `req`.
pub fn EtcdClient::range(
self : EtcdClient,
req : EtcdRangeRequest,
) -> EtcdRangeResponse raise {
EtcdRangeResponse::decode(
etcd_unwrap(
self.channel.call("/etcdserverpb.KV/Range", req.encode()),
"Range",
),
)
}
///|
/// `KV.Put`: store the key/value (optionally under a lease) in `req`.
pub fn EtcdClient::put(
self : EtcdClient,
req : EtcdPutRequest,
) -> EtcdPutResponse raise {
EtcdPutResponse::decode(
etcd_unwrap(self.channel.call("/etcdserverpb.KV/Put", req.encode()), "Put"),
)
}
///|
/// `KV.DeleteRange`: delete the key or range in `req` (deregister).
pub fn EtcdClient::delete_range(
self : EtcdClient,
req : EtcdDeleteRangeRequest,
) -> EtcdDeleteRangeResponse raise {
EtcdDeleteRangeResponse::decode(
etcd_unwrap(
self.channel.call("/etcdserverpb.KV/DeleteRange", req.encode()),
"DeleteRange",
),
)
}
///|
/// `Lease.LeaseGrant`: obtain a lease with the requested TTL.
pub fn EtcdClient::lease_grant(
self : EtcdClient,
req : EtcdLeaseGrantRequest,
) -> EtcdLeaseGrantResponse raise {
EtcdLeaseGrantResponse::decode(
etcd_unwrap(
self.channel.call("/etcdserverpb.Lease/LeaseGrant", req.encode()),
"LeaseGrant",
),
)
}
///|
/// `Watch.Watch`: open a watch with `req` and read the stream of responses the server
/// produces (etcd's Watch is a bidi stream; this drives the common open-then-observe
/// direction over the server-streaming path). Each `WatchResponse` carries the batch
/// of change events since the last one.
pub fn EtcdClient::watch(
self : EtcdClient,
req : EtcdWatchRequest,
) -> Array[EtcdWatchResponse] raise {
let replies = match
self.channel.call_server_streaming(
"/etcdserverpb.Watch/Watch",
req.encode(),
) {
Ok(msgs) => msgs
Err(status) =>
raise EtcdError("Watch failed: " + @moonrpc.Status::name(status))
}
let out : Array[EtcdWatchResponse] = []
for reply in replies {
out.push(EtcdWatchResponse::decode(reply))
}
out
}