//! Asynchronous client & synchronous client.

use crate::channel::Channel;
use crate::error::{Error, Result};
#[cfg(feature = "tls-openssl")]
use crate::openssl_tls::{self, OpenSslClientConfig, OpenSslConnector};
use crate::rpc::auth::Permission;
use crate::rpc::auth::{AuthClient, AuthDisableResponse, AuthEnableResponse};
use crate::rpc::auth::{
    RoleAddResponse, RoleDeleteResponse, RoleGetResponse, RoleGrantPermissionResponse,
    RoleListResponse, RoleRevokePermissionOptions, RoleRevokePermissionResponse, UserAddOptions,
    UserAddResponse, UserChangePasswordResponse, UserDeleteResponse, UserGetResponse,
    UserGrantRoleResponse, UserListResponse, UserRevokeRoleResponse,
};
use crate::rpc::cluster::{
    ClusterClient, MemberAddOptions, MemberAddResponse, MemberListResponse, MemberPromoteResponse,
    MemberRemoveResponse, MemberUpdateResponse,
};
use crate::rpc::election::{
    CampaignResponse, ElectionClient, LeaderResponse, ObserveStream, ProclaimOptions,
    ProclaimResponse, ResignOptions, ResignResponse,
};
use crate::rpc::kv::{
    CompactionOptions, CompactionResponse, DeleteOptions, DeleteResponse, GetOptions, GetResponse,
    KvClient, PutOptions, PutResponse, Txn, TxnResponse,
};
use crate::rpc::lease::{
    LeaseClient, LeaseGrantOptions, LeaseGrantResponse, LeaseKeepAliveStream, LeaseKeeper,
    LeaseLeasesResponse, LeaseRevokeResponse, LeaseTimeToLiveOptions, LeaseTimeToLiveResponse,
};
use crate::rpc::lock::{LockClient, LockOptions, LockResponse, UnlockResponse};
use crate::rpc::maintenance::{
    AlarmAction, AlarmOptions, AlarmResponse, AlarmType, DefragmentResponse, HashKvResponse,
    HashResponse, MaintenanceClient, MoveLeaderResponse, SnapshotStreaming, StatusResponse,
};
use crate::rpc::watch::{WatchClient, WatchOptions, WatchStream, Watcher};
#[cfg(feature = "tls-openssl")]
use crate::OpenSslResult;
#[cfg(feature = "tls")]
use crate::TlsOptions;
use http::uri::Uri;

use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc::Sender;

use tonic::transport::Endpoint;

use tower::discover::Change;

const HTTP_PREFIX: &str = "http://";
const HTTPS_PREFIX: &str = "https://";

/// Asynchronous `etcd` client using v3 API.
#[derive(Clone)]
pub struct Client {
    kv: KvClient,
    watch: WatchClient,
    lease: LeaseClient,
    lock: LockClient,
    auth: AuthClient,
    maintenance: MaintenanceClient,
    cluster: ClusterClient,
    election: ElectionClient,
    options: Option<ConnectOptions>,
    tx: Sender<Change<Uri, Endpoint>>,
}

impl Client {
    /// Connect to `etcd` servers from given `endpoints`.
    pub async fn connect<E: AsRef<str>, S: AsRef<[E]>>(
        endpoints: S,
        options: Option<ConnectOptions>,
    ) -> Result<Self> {
        let endpoints = {
            let mut eps = Vec::new();
            for e in endpoints.as_ref() {
                let channel = Self::build_endpoint(e.as_ref(), &options)?;
                eps.push(channel);
            }
            eps
        };

        if endpoints.is_empty() {
            return Err(Error::InvalidArgs(String::from("empty endpoints")));
        }

        // Always use balance strategy even if there is only one endpoint.
        #[cfg(not(feature = "tls-openssl"))]
        let (channel, tx) = Channel::balance_channel(64);
        #[cfg(feature = "tls-openssl")]
        let (channel, tx) = openssl_tls::balanced_channel(
            options
                .clone()
                .and_then(|o| o.otls)
                .unwrap_or_else(OpenSslConnector::create_default)?,
        )?;
        for endpoint in endpoints {
            // The rx inside `channel` won't be closed or dropped here
            tx.send(Change::Insert(endpoint.uri().clone(), endpoint))
                .await
                .unwrap();
        }

        let mut options = options;
        let auth_token = Self::auth(channel.clone(), &mut options).await?;
        Ok(Self::build_client(channel, tx, auth_token, options))
    }

    fn build_endpoint(url: &str, options: &Option<ConnectOptions>) -> Result<Endpoint> {
        #[cfg(feature = "tls-openssl")]
        use tonic::transport::Channel;
        let mut endpoint = if url.starts_with(HTTP_PREFIX) {
            #[cfg(feature = "tls")]
            if let Some(connect_options) = options {
                if connect_options.tls.is_some() {
                    return Err(Error::InvalidArgs(String::from(
                        "TLS options are only supported with HTTPS URLs",
                    )));
                }
            }

            Channel::builder(url.parse()?)
        } else if url.starts_with(HTTPS_PREFIX) {
            #[cfg(not(any(feature = "tls", feature = "tls-openssl")))]
            return Err(Error::InvalidArgs(String::from(
                "HTTPS URLs are only supported with the feature \"tls\"",
            )));

            #[cfg(all(feature = "tls-openssl", not(feature = "tls")))]
            {
                Channel::builder(url.parse()?)
            }

            #[cfg(feature = "tls")]
            {
                let tls = if let Some(connect_options) = options {
                    connect_options.tls.clone()
                } else {
                    None
                }
                .unwrap_or_else(TlsOptions::new);

                Channel::builder(url.parse()?).tls_config(tls)?
            }
        } else {
            #[cfg(feature = "tls")]
            {
                let tls = if let Some(connect_options) = options {
                    connect_options.tls.clone()
                } else {
                    None
                };

                match tls {
                    Some(tls) => {
                        let e = HTTPS_PREFIX.to_owned() + url;
                        Channel::builder(e.parse()?).tls_config(tls)?
                    }
                    None => {
                        let e = HTTP_PREFIX.to_owned() + url;
                        Channel::builder(e.parse()?)
                    }
                }
            }

            #[cfg(all(feature = "tls-openssl", not(feature = "tls")))]
            {
                let pfx = if options.as_ref().and_then(|o| o.otls.as_ref()).is_some() {
                    HTTPS_PREFIX
                } else {
                    HTTP_PREFIX
                };
                let e = pfx.to_owned() + url;
                Channel::builder(e.parse()?)
            }

            #[cfg(all(not(feature = "tls"), not(feature = "tls-openssl")))]
            {
                let e = HTTP_PREFIX.to_owned() + url;
                Channel::builder(e.parse()?)
            }
        };

        if let Some(opts) = options {
            if let Some((interval, timeout)) = opts.keep_alive {
                endpoint = endpoint
                    .keep_alive_while_idle(opts.keep_alive_while_idle)
                    .http2_keep_alive_interval(interval)
                    .keep_alive_timeout(timeout);
            }

            if let Some(timeout) = opts.timeout {
                endpoint = endpoint.timeout(timeout);
            }

            if let Some(timeout) = opts.connect_timeout {
                endpoint = endpoint.connect_timeout(timeout);
            }
        }

        Ok(endpoint)
    }

    async fn auth(
        channel: Channel,
        options: &mut Option<ConnectOptions>,
    ) -> Result<Option<Arc<http::HeaderValue>>> {
        let user = match options {
            None => return Ok(None),
            Some(opt) => {
                // Take away the user, the password should not be stored in client.
                opt.user.take()
            }
        };

        if let Some((name, password)) = user {
            let mut tmp_auth = AuthClient::new(channel, None);
            let resp = tmp_auth.authenticate(name, password).await?;
            Ok(Some(Arc::new(resp.token().parse()?)))
        } else {
            Ok(None)
        }
    }

    fn build_client(
        channel: Channel,
        tx: Sender<Change<Uri, Endpoint>>,
        auth_token: Option<Arc<http::HeaderValue>>,
        options: Option<ConnectOptions>,
    ) -> Self {
        let kv = KvClient::new(channel.clone(), auth_token.clone());
        let watch = WatchClient::new(channel.clone(), auth_token.clone());
        let lease = LeaseClient::new(channel.clone(), auth_token.clone());
        let lock = LockClient::new(channel.clone(), auth_token.clone());
        let auth = AuthClient::new(channel.clone(), auth_token.clone());
        let cluster = ClusterClient::new(channel.clone(), auth_token.clone());
        let maintenance = MaintenanceClient::new(channel.clone(), auth_token.clone());
        let election = ElectionClient::new(channel, auth_token);

        Self {
            kv,
            watch,
            lease,
            lock,
            auth,
            maintenance,
            cluster,
            election,
            options,
            tx,
        }
    }

    /// Dynamically add an endpoint to the client.
    ///
    /// Which can be used to add a new member to the underlying balance cache.
    /// The typical scenario is that application can use a services discovery
    /// to discover the member list changes and add/remove them to/from the client.
    ///
    /// Note that the [`Client`] doesn't check the authentication before added.
    /// So the etcd member of the added endpoint REQUIRES to use the same auth
    /// token as when create the client. Otherwise, the underlying balance
    /// services will not be able to connect to the new endpoint.
    #[inline]
    pub async fn add_endpoint<E: AsRef<str>>(&self, endpoint: E) -> Result<()> {
        let endpoint = Self::build_endpoint(endpoint.as_ref(), &self.options)?;
        let tx = &self.tx;
        tx.send(Change::Insert(endpoint.uri().clone(), endpoint))
            .await
            .map_err(|e| Error::EndpointError(format!("failed to add endpoint because of {}", e)))
    }

    /// Dynamically remove an endpoint from the client.
    ///
    /// Note that the `endpoint` str should be the same as it was added.
    /// And the underlying balance services cache used the hash from the Uri,
    /// which was parsed from `endpoint` str, to do the equality comparisons.
    #[inline]
    pub async fn remove_endpoint<E: AsRef<str>>(&self, endpoint: E) -> Result<()> {
        let uri = http::Uri::from_str(endpoint.as_ref())?;
        let tx = &self.tx;
        tx.send(Change::Remove(uri)).await.map_err(|e| {
            Error::EndpointError(format!("failed to remove endpoint because of {}", e))
        })
    }

    /// Gets a KV client.
    #[inline]
    pub fn kv_client(&self) -> KvClient {
        self.kv.clone()
    }

    /// Gets a watch client.
    #[inline]
    pub fn watch_client(&self) -> WatchClient {
        self.watch.clone()
    }

    /// Gets a lease client.
    #[inline]
    pub fn lease_client(&self) -> LeaseClient {
        self.lease.clone()
    }

    /// Gets an auth client.
    #[inline]
    pub fn auth_client(&self) -> AuthClient {
        self.auth.clone()
    }

    /// Gets a maintenance client.
    #[inline]
    pub fn maintenance_client(&self) -> MaintenanceClient {
        self.maintenance.clone()
    }

    /// Gets a cluster client.
    #[inline]
    pub fn cluster_client(&self) -> ClusterClient {
        self.cluster.clone()
    }

    /// Gets a lock client.
    #[inline]
    pub fn lock_client(&self) -> LockClient {
        self.lock.clone()
    }

    /// Gets a election client.
    #[inline]
    pub fn election_client(&self) -> ElectionClient {
        self.election.clone()
    }

    /// Put the given key into the key-value store.
    /// A put request increments the revision of the key-value store
    /// and generates one event in the event history.
    #[inline]
    pub async fn put(
        &mut self,
        key: impl Into<Vec<u8>>,
        value: impl Into<Vec<u8>>,
        options: Option<PutOptions>,
    ) -> Result<PutResponse> {
        self.kv.put(key, value, options).await
    }

    /// Gets the key from the key-value store.
    #[inline]
    pub async fn get(
        &mut self,
        key: impl Into<Vec<u8>>,
        options: Option<GetOptions>,
    ) -> Result<GetResponse> {
        self.kv.get(key, options).await
    }

    /// Deletes the given key from the key-value store.
    #[inline]
    pub async fn delete(
        &mut self,
        key: impl Into<Vec<u8>>,
        options: Option<DeleteOptions>,
    ) -> Result<DeleteResponse> {
        self.kv.delete(key, options).await
    }

    /// Compacts the event history in the etcd key-value store. The key-value
    /// store should be periodically compacted or the event history will continue to grow
    /// indefinitely.
    #[inline]
    pub async fn compact(
        &mut self,
        revision: i64,
        options: Option<CompactionOptions>,
    ) -> Result<CompactionResponse> {
        self.kv.compact(revision, options).await
    }

    /// Processes multiple operations in a single transaction.
    /// A txn request increments the revision of the key-value store
    /// and generates events with the same revision for every completed operation.
    /// It is not allowed to modify the same key several times within one txn.
    #[inline]
    pub async fn txn(&mut self, txn: Txn) -> Result<TxnResponse> {
        self.kv.txn(txn).await
    }

    /// Watches for events happening or that have happened. Both input and output
    /// are streams; the input stream is for creating and canceling watcher and the output
    /// stream sends events. The entire event history can be watched starting from the
    /// last compaction revision.
    #[inline]
    pub async fn watch(
        &mut self,
        key: impl Into<Vec<u8>>,
        options: Option<WatchOptions>,
    ) -> Result<(Watcher, WatchStream)> {
        self.watch.watch(key, options).await
    }

    /// Creates a lease which expires if the server does not receive a keepAlive
    /// within a given time to live period. All keys attached to the lease will be expired and
    /// deleted if the lease expires. Each expired key generates a delete event in the event history.
    #[inline]
    pub async fn lease_grant(
        &mut self,
        ttl: i64,
        options: Option<LeaseGrantOptions>,
    ) -> Result<LeaseGrantResponse> {
        self.lease.grant(ttl, options).await
    }

    /// Revokes a lease. All keys attached to the lease will expire and be deleted.
    #[inline]
    pub async fn lease_revoke(&mut self, id: i64) -> Result<LeaseRevokeResponse> {
        self.lease.revoke(id).await
    }

    /// Keeps the lease alive by streaming keep alive requests from the client
    /// to the server and streaming keep alive responses from the server to the client.
    #[inline]
    pub async fn lease_keep_alive(
        &mut self,
        id: i64,
    ) -> Result<(LeaseKeeper, LeaseKeepAliveStream)> {
        self.lease.keep_alive(id).await
    }

    /// Retrieves lease information.
    #[inline]
    pub async fn lease_time_to_live(
        &mut self,
        id: i64,
        options: Option<LeaseTimeToLiveOptions>,
    ) -> Result<LeaseTimeToLiveResponse> {
        self.lease.time_to_live(id, options).await
    }

    /// Lists all existing leases.
    #[inline]
    pub async fn leases(&mut self) -> Result<LeaseLeasesResponse> {
        self.lease.leases().await
    }

    /// Lock acquires a distributed shared lock on a given named lock.
    /// On success, it will return a unique key that exists so long as the
    /// lock is held by the caller. This key can be used in conjunction with
    /// transactions to safely ensure updates to etcd only occur while holding
    /// lock ownership. The lock is held until Unlock is called on the key or the
    /// lease associate with the owner expires.
    #[inline]
    pub async fn lock(
        &mut self,
        name: impl Into<Vec<u8>>,
        options: Option<LockOptions>,
    ) -> Result<LockResponse> {
        self.lock.lock(name, options).await
    }

    /// Unlock takes a key returned by Lock and releases the hold on lock. The
    /// next Lock caller waiting for the lock will then be woken up and given
    /// ownership of the lock.
    #[inline]
    pub async fn unlock(&mut self, key: impl Into<Vec<u8>>) -> Result<UnlockResponse> {
        self.lock.unlock(key).await
    }

    /// Enables authentication.
    #[inline]
    pub async fn auth_enable(&mut self) -> Result<AuthEnableResponse> {
        self.auth.auth_enable().await
    }

    /// Disables authentication.
    #[inline]
    pub async fn auth_disable(&mut self) -> Result<AuthDisableResponse> {
        self.auth.auth_disable().await
    }

    /// Adds role.
    #[inline]
    pub async fn role_add(&mut self, name: impl Into<String>) -> Result<RoleAddResponse> {
        self.auth.role_add(name).await
    }

    /// Deletes role.
    #[inline]
    pub async fn role_delete(&mut self, name: impl Into<String>) -> Result<RoleDeleteResponse> {
        self.auth.role_delete(name).await
    }

    /// Gets role.
    #[inline]
    pub async fn role_get(&mut self, name: impl Into<String>) -> Result<RoleGetResponse> {
        self.auth.role_get(name).await
    }

    /// Lists role.
    #[inline]
    pub async fn role_list(&mut self) -> Result<RoleListResponse> {
        self.auth.role_list().await
    }

    /// Grants role permission.
    #[inline]
    pub async fn role_grant_permission(
        &mut self,
        name: impl Into<String>,
        perm: Permission,
    ) -> Result<RoleGrantPermissionResponse> {
        self.auth.role_grant_permission(name, perm).await
    }

    /// Revokes role permission.
    #[inline]
    pub async fn role_revoke_permission(
        &mut self,
        name: impl Into<String>,
        key: impl Into<Vec<u8>>,
        options: Option<RoleRevokePermissionOptions>,
    ) -> Result<RoleRevokePermissionResponse> {
        self.auth.role_revoke_permission(name, key, options).await
    }

    /// Add an user.
    #[inline]
    pub async fn user_add(
        &mut self,
        name: impl Into<String>,
        password: impl Into<String>,
        options: Option<UserAddOptions>,
    ) -> Result<UserAddResponse> {
        self.auth.user_add(name, password, options).await
    }

    /// Gets the user info by the user name.
    #[inline]
    pub async fn user_get(&mut self, name: impl Into<String>) -> Result<UserGetResponse> {
        self.auth.user_get(name).await
    }

    /// Lists all users.
    #[inline]
    pub async fn user_list(&mut self) -> Result<UserListResponse> {
        self.auth.user_list().await
    }

    /// Deletes the given key from the key-value store.
    #[inline]
    pub async fn user_delete(&mut self, name: impl Into<String>) -> Result<UserDeleteResponse> {
        self.auth.user_delete(name).await
    }

    /// Change password for an user.
    #[inline]
    pub async fn user_change_password(
        &mut self,
        name: impl Into<String>,
        password: impl Into<String>,
    ) -> Result<UserChangePasswordResponse> {
        self.auth.user_change_password(name, password).await
    }

    /// Grant role for an user.
    #[inline]
    pub async fn user_grant_role(
        &mut self,
        user: impl Into<String>,
        role: impl Into<String>,
    ) -> Result<UserGrantRoleResponse> {
        self.auth.user_grant_role(user, role).await
    }

    /// Revoke role for an user.
    #[inline]
    pub async fn user_revoke_role(
        &mut self,
        user: impl Into<String>,
        role: impl Into<String>,
    ) -> Result<UserRevokeRoleResponse> {
        self.auth.user_revoke_role(user, role).await
    }

    /// Maintain(get, active or inactive) alarms of members.
    #[inline]
    pub async fn alarm(
        &mut self,
        alarm_action: AlarmAction,
        alarm_type: AlarmType,
        options: Option<AlarmOptions>,
    ) -> Result<AlarmResponse> {
        self.maintenance
            .alarm(alarm_action, alarm_type, options)
            .await
    }

    /// Gets the status of a member.
    #[inline]
    pub async fn status(&mut self) -> Result<StatusResponse> {
        self.maintenance.status().await
    }

    /// Defragments a member's backend database to recover storage space.
    #[inline]
    pub async fn defragment(&mut self) -> Result<DefragmentResponse> {
        self.maintenance.defragment().await
    }

    /// Computes the hash of whole backend keyspace.
    /// including key, lease, and other buckets in storage.
    /// This is designed for testing ONLY!
    #[inline]
    pub async fn hash(&mut self) -> Result<HashResponse> {
        self.maintenance.hash().await
    }

    /// Computes the hash of all MVCC keys up to a given revision.
    /// It only iterates \"key\" bucket in backend storage.
    #[inline]
    pub async fn hash_kv(&mut self, revision: i64) -> Result<HashKvResponse> {
        self.maintenance.hash_kv(revision).await
    }

    /// Gets a snapshot of the entire backend from a member over a stream to a client.
    #[inline]
    pub async fn snapshot(&mut self) -> Result<SnapshotStreaming> {
        self.maintenance.snapshot().await
    }

    /// Adds current connected server as a member.
    #[inline]
    pub async fn member_add<E: AsRef<str>, S: AsRef<[E]>>(
        &mut self,
        urls: S,
        options: Option<MemberAddOptions>,
    ) -> Result<MemberAddResponse> {
        let mut eps = Vec::new();
        for e in urls.as_ref() {
            let e = e.as_ref();
            let url = if e.starts_with(HTTP_PREFIX) || e.starts_with(HTTPS_PREFIX) {
                e.to_string()
            } else {
                HTTP_PREFIX.to_owned() + e
            };
            eps.push(url);
        }

        self.cluster.member_add(eps, options).await
    }

    /// Remove a member.
    #[inline]
    pub async fn member_remove(&mut self, id: u64) -> Result<MemberRemoveResponse> {
        self.cluster.member_remove(id).await
    }

    /// Updates the member.
    #[inline]
    pub async fn member_update(
        &mut self,
        id: u64,
        url: impl Into<Vec<String>>,
    ) -> Result<MemberUpdateResponse> {
        self.cluster.member_update(id, url).await
    }

    /// Promotes the member.
    #[inline]
    pub async fn member_promote(&mut self, id: u64) -> Result<MemberPromoteResponse> {
        self.cluster.member_promote(id).await
    }

    /// Lists members.
    #[inline]
    pub async fn member_list(&mut self) -> Result<MemberListResponse> {
        self.cluster.member_list().await
    }

    /// Moves the current leader node to target node.
    #[inline]
    pub async fn move_leader(&mut self, target_id: u64) -> Result<MoveLeaderResponse> {
        self.maintenance.move_leader(target_id).await
    }

    /// Puts a value as eligible for the election on the prefix key.
    /// Multiple sessions can participate in the election for the
    /// same prefix, but only one can be the leader at a time.
    #[inline]
    pub async fn campaign(
        &mut self,
        name: impl Into<Vec<u8>>,
        value: impl Into<Vec<u8>>,
        lease: i64,
    ) -> Result<CampaignResponse> {
        self.election.campaign(name, value, lease).await
    }

    /// Lets the leader announce a new value without another election.
    #[inline]
    pub async fn proclaim(
        &mut self,
        value: impl Into<Vec<u8>>,
        options: Option<ProclaimOptions>,
    ) -> Result<ProclaimResponse> {
        self.election.proclaim(value, options).await
    }

    /// Returns the leader value for the current election.
    #[inline]
    pub async fn leader(&mut self, name: impl Into<Vec<u8>>) -> Result<LeaderResponse> {
        self.election.leader(name).await
    }

    /// Returns a channel that reliably observes ordered leader proposals
    /// as GetResponse values on every current elected leader key.
    #[inline]
    pub async fn observe(&mut self, name: impl Into<Vec<u8>>) -> Result<ObserveStream> {
        self.election.observe(name).await
    }

    /// Releases election leadership and then start a new election
    #[inline]
    pub async fn resign(&mut self, option: Option<ResignOptions>) -> Result<ResignResponse> {
        self.election.resign(option).await
    }
}

/// Options for `Connect` operation.
#[derive(Debug, Default, Clone)]
pub struct ConnectOptions {
    /// user is a pair values of name and password
    user: Option<(String, String)>,
    /// HTTP2 keep-alive: (keep_alive_interval, keep_alive_timeout)
    keep_alive: Option<(Duration, Duration)>,
    /// Whether send keep alive pings even there are no active streams.
    keep_alive_while_idle: bool,
    /// Apply a timeout to each gRPC request.
    timeout: Option<Duration>,
    /// Apply a timeout to connecting to the endpoint.
    connect_timeout: Option<Duration>,
    #[cfg(feature = "tls")]
    tls: Option<TlsOptions>,
    #[cfg(feature = "tls-openssl")]
    otls: Option<OpenSslResult<OpenSslConnector>>,
}

impl ConnectOptions {
    /// name is the identifier for the distributed shared lock to be acquired.
    #[inline]
    pub fn with_user(mut self, name: impl Into<String>, password: impl Into<String>) -> Self {
        self.user = Some((name.into(), password.into()));
        self
    }

    /// Sets TLS options.
    ///
    /// Notes that this function have to work with `HTTPS` URLs.
    #[cfg_attr(docsrs, doc(cfg(feature = "tls")))]
    #[cfg(feature = "tls")]
    #[inline]
    pub fn with_tls(mut self, tls: TlsOptions) -> Self {
        self.tls = Some(tls);
        self
    }

    /// Sets TLS options, however using the OpenSSL implementation.
    #[cfg_attr(docsrs, doc(cfg(feature = "tls-openssl")))]
    #[cfg(feature = "tls-openssl")]
    #[inline]
    pub fn with_openssl_tls(mut self, otls: OpenSslClientConfig) -> Self {
        // NOTE1: Perhaps we can unify the essential TLS config terms by something like `TlsBuilder`?
        //
        // NOTE2: we delay the checking at connection step to keep consistency with tonic, however would
        // things be better if we validate the config at here?
        self.otls = Some(otls.build());
        self
    }

    /// Enable HTTP2 keep-alive with `interval` and `timeout`.
    #[inline]
    pub fn with_keep_alive(mut self, interval: Duration, timeout: Duration) -> Self {
        self.keep_alive = Some((interval, timeout));
        self
    }

    /// Apply a timeout to each request.
    #[inline]
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(timeout);
        self
    }

    /// Apply a timeout to connecting to the endpoint.
    #[inline]
    pub fn with_connect_timeout(mut self, timeout: Duration) -> Self {
        self.connect_timeout = Some(timeout);
        self
    }

    /// Whether send keep alive pings even there are no active requests.
    /// If disabled, keep-alive pings are only sent while there are opened request/response streams.
    /// If enabled, pings are also sent when no streams are active.
    /// NOTE: Some implementations of gRPC server may send GOAWAY if there are too many pings.
    ///       This would be useful if you meet some error like `too many pings`.
    #[inline]
    pub fn with_keep_alive_while_idle(mut self, enabled: bool) -> Self {
        self.keep_alive_while_idle = enabled;
        self
    }

    /// Creates a `ConnectOptions`.
    #[inline]
    pub const fn new() -> Self {
        ConnectOptions {
            user: None,
            keep_alive: None,
            keep_alive_while_idle: true,
            timeout: None,
            connect_timeout: None,
            #[cfg(feature = "tls")]
            tls: None,
            #[cfg(feature = "tls-openssl")]
            otls: None,
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Compare, CompareOp, EventType, PermissionType, TxnOp, TxnOpResponse};

    const DEFAULT_TEST_ENDPOINT: &str = "localhost:2379";

    /// Get client for testing.
    async fn get_client() -> Result<Client> {
        Client::connect([DEFAULT_TEST_ENDPOINT], None).await
    }

    #[tokio::test]
    async fn test_put() -> Result<()> {
        let mut client = get_client().await?;
        client.put("put", "123", None).await?;

        // overwrite with prev key
        {
            let resp = client
                .put("put", "456", Some(PutOptions::new().with_prev_key()))
                .await?;
            let prev_key = resp.prev_key();
            assert!(prev_key.is_some());
            let prev_key = prev_key.unwrap();
            assert_eq!(prev_key.key(), b"put");
            assert_eq!(prev_key.value(), b"123");
        }

        // overwrite again with prev key
        {
            let resp = client
                .put("put", "789", Some(PutOptions::new().with_prev_key()))
                .await?;
            let prev_key = resp.prev_key();
            assert!(prev_key.is_some());
            let prev_key = prev_key.unwrap();
            assert_eq!(prev_key.key(), b"put");
            assert_eq!(prev_key.value(), b"456");
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_get() -> Result<()> {
        let mut client = get_client().await?;
        client.put("get10", "10", None).await?;
        client.put("get11", "11", None).await?;
        client.put("get20", "20", None).await?;
        client.put("get21", "21", None).await?;

        // get key
        {
            let resp = client.get("get11", None).await?;
            assert_eq!(resp.count(), 1);
            assert!(!resp.more());
            assert_eq!(resp.kvs().len(), 1);
            assert_eq!(resp.kvs()[0].key(), b"get11");
            assert_eq!(resp.kvs()[0].value(), b"11");
        }

        // get from key
        {
            let resp = client
                .get(
                    "get11",
                    Some(GetOptions::new().with_from_key().with_limit(2)),
                )
                .await?;
            assert!(resp.more());
            assert_eq!(resp.kvs().len(), 2);
            assert_eq!(resp.kvs()[0].key(), b"get11");
            assert_eq!(resp.kvs()[0].value(), b"11");
            assert_eq!(resp.kvs()[1].key(), b"get20");
            assert_eq!(resp.kvs()[1].value(), b"20");
        }

        // get prefix keys
        {
            let resp = client
                .get("get1", Some(GetOptions::new().with_prefix()))
                .await?;
            assert_eq!(resp.count(), 2);
            assert!(!resp.more());
            assert_eq!(resp.kvs().len(), 2);
            assert_eq!(resp.kvs()[0].key(), b"get10");
            assert_eq!(resp.kvs()[0].value(), b"10");
            assert_eq!(resp.kvs()[1].key(), b"get11");
            assert_eq!(resp.kvs()[1].value(), b"11");
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_delete() -> Result<()> {
        let mut client = get_client().await?;
        client.put("del10", "10", None).await?;
        client.put("del11", "11", None).await?;
        client.put("del20", "20", None).await?;
        client.put("del21", "21", None).await?;
        client.put("del31", "31", None).await?;
        client.put("del32", "32", None).await?;

        // delete key
        {
            let resp = client.delete("del11", None).await?;
            assert_eq!(resp.deleted(), 1);
            let resp = client
                .get("del11", Some(GetOptions::new().with_count_only()))
                .await?;
            assert_eq!(resp.count(), 0);
        }

        // delete a range of keys
        {
            let resp = client
                .delete("del11", Some(DeleteOptions::new().with_range("del22")))
                .await?;
            assert_eq!(resp.deleted(), 2);
            let resp = client
                .get(
                    "del11",
                    Some(GetOptions::new().with_range("del22").with_count_only()),
                )
                .await?;
            assert_eq!(resp.count(), 0);
        }

        // delete key with prefix
        {
            let resp = client
                .delete("del3", Some(DeleteOptions::new().with_prefix()))
                .await?;
            assert_eq!(resp.deleted(), 2);
            let resp = client
                .get("del32", Some(GetOptions::new().with_count_only()))
                .await?;
            assert_eq!(resp.count(), 0);
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_compact() -> Result<()> {
        let mut client = get_client().await?;
        let rev0 = client
            .put("compact", "0", None)
            .await?
            .header()
            .unwrap()
            .revision();
        let rev1 = client
            .put("compact", "1", None)
            .await?
            .header()
            .unwrap()
            .revision();

        // before compacting
        let rev0_resp = client
            .get("compact", Some(GetOptions::new().with_revision(rev0)))
            .await?;
        assert_eq!(rev0_resp.kvs()[0].value(), b"0");
        let rev1_resp = client
            .get("compact", Some(GetOptions::new().with_revision(rev1)))
            .await?;
        assert_eq!(rev1_resp.kvs()[0].value(), b"1");

        client.compact(rev1, None).await?;

        // after compacting
        let result = client
            .get("compact", Some(GetOptions::new().with_revision(rev0)))
            .await;
        assert!(result.is_err());
        let rev1_resp = client
            .get("compact", Some(GetOptions::new().with_revision(rev1)))
            .await?;
        assert_eq!(rev1_resp.kvs()[0].value(), b"1");

        Ok(())
    }

    #[tokio::test]
    async fn test_txn() -> Result<()> {
        let mut client = get_client().await?;
        client.put("txn01", "01", None).await?;

        // transaction 1
        {
            let resp = client
                .txn(
                    Txn::new()
                        .when(&[Compare::value("txn01", CompareOp::Equal, "01")][..])
                        .and_then(
                            &[TxnOp::put(
                                "txn01",
                                "02",
                                Some(PutOptions::new().with_prev_key()),
                            )][..],
                        )
                        .or_else(&[TxnOp::get("txn01", None)][..]),
                )
                .await?;

            assert!(resp.succeeded());
            let op_responses = resp.op_responses();
            assert_eq!(op_responses.len(), 1);

            match op_responses[0] {
                TxnOpResponse::Put(ref resp) => assert_eq!(resp.prev_key().unwrap().value(), b"01"),
                _ => panic!("unexpected response"),
            }

            let resp = client.get("txn01", None).await?;
            assert_eq!(resp.kvs()[0].key(), b"txn01");
            assert_eq!(resp.kvs()[0].value(), b"02");
        }

        // transaction 2
        {
            let resp = client
                .txn(
                    Txn::new()
                        .when(&[Compare::value("txn01", CompareOp::Equal, "01")][..])
                        .and_then(&[TxnOp::put("txn01", "02", None)][..])
                        .or_else(&[TxnOp::get("txn01", None)][..]),
                )
                .await?;

            assert!(!resp.succeeded());
            let op_responses = resp.op_responses();
            assert_eq!(op_responses.len(), 1);

            match op_responses[0] {
                TxnOpResponse::Get(ref resp) => assert_eq!(resp.kvs()[0].value(), b"02"),
                _ => panic!("unexpected response"),
            }
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_watch() -> Result<()> {
        let mut client = get_client().await?;

        let (mut watcher, mut stream) = client.watch("watch01", None).await?;

        client.put("watch01", "01", None).await?;

        let resp = stream.message().await?.unwrap();
        assert_eq!(resp.watch_id(), watcher.watch_id());
        assert_eq!(resp.events().len(), 1);

        let kv = resp.events()[0].kv().unwrap();
        assert_eq!(kv.key(), b"watch01");
        assert_eq!(kv.value(), b"01");
        assert_eq!(resp.events()[0].event_type(), EventType::Put);

        watcher.cancel().await?;

        let resp = stream.message().await?.unwrap();
        assert_eq!(resp.watch_id(), watcher.watch_id());
        assert!(resp.canceled());

        Ok(())
    }

    #[tokio::test]
    async fn test_grant_revoke() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.lease_grant(123, None).await?;
        assert_eq!(resp.ttl(), 123);
        let id = resp.id();
        client.lease_revoke(id).await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_keep_alive() -> Result<()> {
        let mut client = get_client().await?;

        let resp = client.lease_grant(60, None).await?;
        assert_eq!(resp.ttl(), 60);
        let id = resp.id();

        let (mut keeper, mut stream) = client.lease_keep_alive(id).await?;
        keeper.keep_alive().await?;

        let resp = stream.message().await?.unwrap();
        assert_eq!(resp.id(), keeper.id());
        assert_eq!(resp.ttl(), 60);

        client.lease_revoke(id).await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_time_to_live() -> Result<()> {
        let mut client = get_client().await?;
        let leaseid = 200;
        let resp = client
            .lease_grant(60, Some(LeaseGrantOptions::new().with_id(leaseid)))
            .await?;
        assert_eq!(resp.ttl(), 60);
        assert_eq!(resp.id(), leaseid);

        let resp = client.lease_time_to_live(leaseid, None).await?;
        assert_eq!(resp.id(), leaseid);
        assert_eq!(resp.granted_ttl(), 60);

        client.lease_revoke(leaseid).await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_leases() -> Result<()> {
        let lease1 = 100;
        let lease2 = 101;
        let lease3 = 102;

        let mut client = get_client().await?;
        let resp = client
            .lease_grant(60, Some(LeaseGrantOptions::new().with_id(lease1)))
            .await?;
        assert_eq!(resp.ttl(), 60);
        assert_eq!(resp.id(), lease1);

        let resp = client
            .lease_grant(60, Some(LeaseGrantOptions::new().with_id(lease2)))
            .await?;
        assert_eq!(resp.ttl(), 60);
        assert_eq!(resp.id(), lease2);

        let resp = client
            .lease_grant(60, Some(LeaseGrantOptions::new().with_id(lease3)))
            .await?;
        assert_eq!(resp.ttl(), 60);
        assert_eq!(resp.id(), lease3);

        let resp = client.leases().await?;
        let leases: Vec<_> = resp.leases().iter().map(|status| status.id()).collect();
        assert!(leases.contains(&lease1));
        assert!(leases.contains(&lease2));
        assert!(leases.contains(&lease3));

        client.lease_revoke(lease1).await?;
        client.lease_revoke(lease2).await?;
        client.lease_revoke(lease3).await?;
        Ok(())
    }

    #[tokio::test]
    async fn test_lock() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.lock("lock-test", None).await?;
        let key = resp.key();
        let key_str = std::str::from_utf8(key)?;
        assert!(key_str.starts_with("lock-test/"));

        client.unlock(key).await?;
        Ok(())
    }

    #[ignore]
    #[tokio::test]
    async fn test_auth() -> Result<()> {
        let mut client = get_client().await?;
        client.auth_enable().await?;

        // after enable auth, must operate by authenticated client
        client.put("auth-test", "value", None).await.unwrap_err();

        // connect with authenticate, the user must already exists
        let options = Some(ConnectOptions::new().with_user(
            "root",    // user name
            "rootpwd", // password
        ));
        let mut client_auth = Client::connect(["localhost:2379"], options).await?;
        client_auth.put("auth-test", "value", None).await?;

        client_auth.auth_disable().await?;

        // after disable auth, operate ok
        let mut client = get_client().await?;
        client.put("auth-test", "value", None).await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_role() -> Result<()> {
        let mut client = get_client().await?;

        let role1 = "role1";
        let role2 = "role2";

        let _ = client.role_delete(role1).await;
        let _ = client.role_delete(role2).await;

        client.role_add(role1).await?;

        client.role_get(role1).await?;

        client.role_delete(role1).await?;
        client.role_get(role1).await.unwrap_err();

        client.role_add(role2).await?;
        client.role_get(role2).await?;

        {
            let resp = client.role_list().await?;
            assert!(resp.roles().contains(&role2.to_string()));
        }

        client
            .role_grant_permission(role2, Permission::read("123"))
            .await?;
        client
            .role_grant_permission(role2, Permission::write("abc").with_from_key())
            .await?;
        client
            .role_grant_permission(role2, Permission::read_write("hi").with_range_end("hjj"))
            .await?;
        client
            .role_grant_permission(
                role2,
                Permission::new(PermissionType::Write, "pp").with_prefix(),
            )
            .await?;
        client
            .role_grant_permission(
                role2,
                Permission::new(PermissionType::Read, "xyz").with_all_keys(),
            )
            .await?;

        {
            let resp = client.role_get(role2).await?;
            let permissions = resp.permissions();
            assert!(permissions.contains(&Permission::read("123")));
            assert!(permissions.contains(&Permission::write("abc").with_from_key()));
            assert!(permissions.contains(&Permission::read_write("hi").with_range_end("hjj")));
            assert!(permissions.contains(&Permission::write("pp").with_prefix()));
            assert!(permissions.contains(&Permission::read("xyz").with_all_keys()));
        }

        //revoke all permission
        client.role_revoke_permission(role2, "123", None).await?;
        client
            .role_revoke_permission(
                role2,
                "abc",
                Some(RoleRevokePermissionOptions::new().with_from_key()),
            )
            .await?;
        client
            .role_revoke_permission(
                role2,
                "hi",
                Some(RoleRevokePermissionOptions::new().with_range_end("hjj")),
            )
            .await?;
        client
            .role_revoke_permission(
                role2,
                "pp",
                Some(RoleRevokePermissionOptions::new().with_prefix()),
            )
            .await?;
        client
            .role_revoke_permission(
                role2,
                "xyz",
                Some(RoleRevokePermissionOptions::new().with_all_keys()),
            )
            .await?;

        let resp = client.role_get(role2).await?;
        assert!(resp.permissions().is_empty());

        client.role_delete(role2).await?;

        Ok(())
    }

    #[tokio::test]
    async fn test_user() -> Result<()> {
        let name1 = "usr1";
        let password1 = "pwd1";
        let name2 = "usr2";
        let password2 = "pwd2";
        let name3 = "usr3";
        let password3 = "pwd3";
        let role1 = "role1";

        let mut client = get_client().await?;

        // ignore result
        let _resp = client.user_delete(name1).await;
        let _resp = client.user_delete(name2).await;
        let _resp = client.user_delete(name3).await;
        let _resp = client.role_delete(role1).await;

        client
            .user_add(name1, password1, Some(UserAddOptions::new()))
            .await?;

        client
            .user_add(name2, password2, Some(UserAddOptions::new().with_no_pwd()))
            .await?;

        client.user_add(name3, password3, None).await?;

        client.user_get(name1).await?;

        {
            let resp = client.user_list().await?;
            assert!(resp.users().contains(&name1.to_string()));
        }

        client.user_delete(name2).await?;
        client.user_get(name2).await.unwrap_err();

        client.user_change_password(name1, password2).await?;
        client.user_get(name1).await?;

        client.role_add(role1).await?;
        client.user_grant_role(name1, role1).await?;
        client.user_get(name1).await?;

        client.user_revoke_role(name1, role1).await?;
        client.user_get(name1).await?;

        let _ = client.user_delete(name1).await;
        let _ = client.user_delete(name2).await;
        let _ = client.user_delete(name3).await;
        let _ = client.role_delete(role1).await;

        Ok(())
    }

    #[tokio::test]
    async fn test_alarm() -> Result<()> {
        let mut client = get_client().await?;

        // Test deactivate alarm.
        {
            let options = AlarmOptions::new();
            let _resp = client
                .alarm(AlarmAction::Deactivate, AlarmType::None, Some(options))
                .await?;
        }

        // Test get None alarm.
        let member_id = {
            let resp = client
                .alarm(AlarmAction::Get, AlarmType::None, None)
                .await?;
            let mems = resp.alarms();
            assert_eq!(mems.len(), 0);
            0
        };

        let mut options = AlarmOptions::new();
        options.with_member(member_id);

        // Test get no space alarm.
        {
            let resp = client
                .alarm(AlarmAction::Get, AlarmType::Nospace, Some(options.clone()))
                .await?;
            let mems = resp.alarms();
            assert_eq!(mems.len(), 0);
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_status() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.status().await?;

        let db_size = resp.db_size();
        assert_ne!(db_size, 0);
        Ok(())
    }

    #[tokio::test]
    async fn test_defragment() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.defragment().await?;
        let hd = resp.header();
        assert!(hd.is_none());
        Ok(())
    }

    #[tokio::test]
    async fn test_hash() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.hash().await?;
        let hd = resp.header();
        assert!(hd.is_some());
        assert_ne!(resp.hash(), 0);
        Ok(())
    }

    #[tokio::test]
    async fn test_hash_kv() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.hash_kv(0).await?;
        let hd = resp.header();
        assert!(hd.is_some());
        assert_ne!(resp.hash(), 0);
        assert_ne!(resp.compact_version(), 0);
        Ok(())
    }

    #[tokio::test]
    async fn test_snapshot() -> Result<()> {
        let mut client = get_client().await?;
        let mut msg = client.snapshot().await?;
        loop {
            if let Some(resp) = msg.message().await? {
                assert!(!resp.blob().is_empty());
                if resp.remaining_bytes() == 0 {
                    break;
                }
            }
        }
        Ok(())
    }

    #[ignore]
    #[tokio::test]
    async fn test_cluster() -> Result<()> {
        let node1 = "localhost:2520";
        let node2 = "localhost:2530";
        let node3 = "localhost:2540";
        let mut client = get_client().await?;
        let resp = client
            .member_add([node1], Some(MemberAddOptions::new().with_is_learner()))
            .await?;
        let id1 = resp.member().unwrap().id();

        let resp = client.member_add([node2], None).await?;
        let id2 = resp.member().unwrap().id();
        let resp = client.member_add([node3], None).await?;
        let id3 = resp.member().unwrap().id();

        let resp = client.member_list().await?;
        let members: Vec<_> = resp.members().iter().map(|member| member.id()).collect();
        assert!(members.contains(&id1));
        assert!(members.contains(&id2));
        assert!(members.contains(&id3));
        Ok(())
    }

    #[tokio::test]
    async fn test_move_leader() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.member_list().await?;
        let member_list = resp.members();

        let resp = client.status().await?;
        let leader_id = resp.leader();
        println!("status {:?}, leader_id {:?}", resp, resp.leader());

        let mut member_id = leader_id;
        for member in member_list {
            println!("member_id {:?}, name is {:?}", member.id(), member.name());
            if member.id() != leader_id {
                member_id = member.id();
                break;
            }
        }

        let resp = client.move_leader(member_id).await?;
        let header = resp.header();
        if member_id == leader_id {
            assert!(header.is_none());
        } else {
            assert!(header.is_some());
        }

        Ok(())
    }

    #[tokio::test]
    async fn test_election() -> Result<()> {
        let mut client = get_client().await?;
        let resp = client.lease_grant(10, None).await?;
        let lease_id = resp.id();
        assert_eq!(resp.ttl(), 10);

        let resp = client.campaign("myElection", "123", lease_id).await?;
        let leader = resp.leader().unwrap();
        assert_eq!(leader.name(), b"myElection");
        assert_eq!(leader.lease(), lease_id);

        let resp = client
            .proclaim(
                "123",
                Some(ProclaimOptions::new().with_leader(leader.clone())),
            )
            .await?;
        let header = resp.header();
        println!("proclaim header {:?}", header.unwrap());
        assert!(header.is_some());

        let mut msg = client.observe(leader.name()).await?;
        loop {
            if let Some(resp) = msg.message().await? {
                assert!(resp.kv().is_some());
                println!("observe key {:?}", resp.kv().unwrap().key_str());
                if resp.kv().is_some() {
                    break;
                }
            }
        }

        let resp = client.leader("myElection").await?;
        let kv = resp.kv().unwrap();
        assert_eq!(kv.value(), b"123");
        assert_eq!(kv.key(), leader.key());
        println!("key is {:?}", kv.key_str());
        println!("value is {:?}", kv.value_str());

        let resign_option = ResignOptions::new().with_leader(leader.clone());

        let resp = client.resign(Some(resign_option)).await?;
        let header = resp.header();
        println!("resign header {:?}", header.unwrap());
        assert!(header.is_some());

        Ok(())
    }

    #[tokio::test]
    async fn test_remove_and_add_endpoint() -> Result<()> {
        let mut client = get_client().await?;
        client.put("endpoint", "add_remove", None).await?;

        // get key
        {
            let resp = client.get("endpoint", None).await?;
            assert_eq!(resp.count(), 1);
            assert!(!resp.more());
            assert_eq!(resp.kvs().len(), 1);
            assert_eq!(resp.kvs()[0].key(), b"endpoint");
            assert_eq!(resp.kvs()[0].value(), b"add_remove");
        }

        // remove endpoint
        client.remove_endpoint(DEFAULT_TEST_ENDPOINT).await?;
        // `Client::get` will hang before adding the endpoint back
        client.add_endpoint(DEFAULT_TEST_ENDPOINT).await?;

        // get key after remove and add endpoint
        {
            let resp = client.get("endpoint", None).await?;
            assert_eq!(resp.count(), 1);
            assert!(!resp.more());
            assert_eq!(resp.kvs().len(), 1);
            assert_eq!(resp.kvs()[0].key(), b"endpoint");
            assert_eq!(resp.kvs()[0].value(), b"add_remove");
        }

        Ok(())
    }
}