fn new(
        client: KvClient,
        watch_client: &WatchClient,
        cache_size: usize,
        cache_enable: bool,
    ) -> Arc<Self> {
        let etcd_cache_size = if cache_size == 0 {
            ETCD_CACHE_DEFAULT_SIZE
        } else {
            cache_size
        };

        let (watch_req_sender, watch_req_receiver) = unbounded::<EtcdWatchRequest>();
        let (watch_id_sender, watch_id_receiver) = bounded::<i64>(1);
        let (shutdown_tx, shutdown_rx) = unbounded();
        let shutdown_response = shutdown_rx.clone();

        let cache = Cache::new(etcd_cache_size, watch_req_sender.clone());
        let (mut client_req_sender, mut client_resp_receiver) = watch_client
            .watch()
            .unwrap_or_else(|e| panic!("failed to send watch commend, the response is: {}", e));

        let cache_clone = cache.clone();
        let cache_inner = cache.clone();

        let this = Arc::new(Self {
            client,
            cache,
            cache_enable,
            watch_sender: watch_req_sender,
            cache_availability: RwLock::new(()),
            shutdown: shutdown_tx,
        });

        let this_clone = Arc::<Self>::clone(&this);
        let this_clone2 = Arc::<Self>::clone(&this);

        if cache_enable {
            // Task that handles all the pending watch requests.
            smol::spawn(async move {
                let mut shutdown_rx = shutdown_rx.into_future().fuse();
                while let Ok(watch_req) = watch_req_receiver.recv().await {
                    let processing_key = watch_req.get_key();
                    futures::select! {
                        res = client_req_sender.send((watch_req.clone().into(), WriteFlags::default())).fuse() => {
                            if let Err(e) = res {
                                warn!("Fail to send watch request, the error is: {}, clean cache", e);
                                this_clone.clean_cache().await;
                                continue;
                            }
                        },
                        _ = shutdown_rx => return
                    };
                    // Wait until etcd server returns watch id.
                    let watch_id = match watch_id_receiver.recv().await {
                        Err(e) => {
                            warn!("Fail to receive watch id from channel, the error is {}, clean cache", e);
                            this_clone.clean_cache().await;
                            continue;
                        }
                        Ok(id) => id,
                    };
                    // Watch request can only be create or cancel.
                    if watch_req.is_create() {
                        cache_clone.insert_watch_id(processing_key.clone(), watch_id);
                    } else {
                        cache_clone.delete_watch_id(&processing_key);
                    }
                }
            })
            .detach();

            // Task that handle the watch responses from Etcd server.
            smol::spawn(async move {
                let mut shutdown_rx = shutdown_response.into_future().fuse();
                loop {
                    let watch_resp = futures::select! {
                        resp_opt = client_resp_receiver.next().fuse() => {
                            if let Some(resp) = resp_opt {
                                resp
                            } else {
                                break;
                            }
                        },
                        _ = shutdown_rx => return
                    };

                    match watch_resp {
                        Ok(resp) => {
                            // TODO: Check if need to spawn new task here.
                            if resp.get_created() || resp.get_canceled() {
                                if let Err(e) = watch_id_sender.send(resp.get_watch_id()).await {
                                    warn!("Fail to send watch id, the error is {}, clean cache", e);
                                    this_clone2.clean_cache().await;
                                    continue;
                                }
                            } else {
                                let events = resp.get_events().to_vec();
                                for event in events {
                                    if event.get_field_type() == Event_EventType::PUT {
                                        if let Some(valid_value) = cache_inner
                                            .search(event.get_kv().get_key().to_vec())
                                            .await
                                        {
                                            // Only update the cache if event's version is larger than
                                            // existing value's version
                                            if valid_value.version < event.get_kv().get_version() {
                                                cache_inner
                                                    .insert(
                                                        event.get_kv().get_key().to_vec(),
                                                        event.get_kv().clone(),
                                                    )
                                                    .await;
                                            }
                                        }
                                    } else {
                                        cache_inner
                                            .delete(event.get_kv().get_key().to_vec(), true)
                                            .await;
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            warn!(
                                "Watch response contains error, the error is: {}, clean cache",
                                e
                            );
                            this_clone2.clean_cache().await;
                            continue;
                        }
                    }
                }
            })
            .detach();
        };
        this
    }
// make no sense to split new()
make no sense to split new()
make no sense to split new ()
fn new(
        client: KvClient,
        watch_client: &WatchClient,
        cache_size: usize,
        cache_enable: bool,
    ) -> Arc<Self> {
        let etcd_cache_size = if cache_size == 0 {
            ETCD_CACHE_DEFAULT_SIZE
        } else {
            cache_size
        };
        let (watch_req_sender, watch_req_receiver) = unbounded::<EtcdWatchRequest>();
        let (watch_id_sender, watch_id_receiver) = bounded::<i64>(1);
        let (shutdown_tx, shutdown_rx) = unbounded();
        let shutdown_response = shutdown_rx.clone();
        let cache = Cache::new(etcd_cache_size, watch_req_sender.clone());
        let (mut client_req_sender, mut client_resp_receiver) = watch_client
            .watch()
            .unwrap_or_else(|e| panic!("failed to send watch commend, the response is: {}", e));
        let cache_clone = cache.clone();
        let cache_inner = cache.clone();
        let this = Arc::new(Self {
            client,
            cache,
            cache_enable,
            watch_sender: watch_req_sender,
            cache_availability: RwLock::new(()),
            shutdown: shutdown_tx,
        });
        let this_clone = Arc::<Self>::clone(&this);
        let this_clone2 = Arc::<Self>::clone(&this);
        if cache_enable {
            // Task that handles all the pending watch requests.
            smol::spawn(async move {
                let mut shutdown_rx = shutdown_rx.into_future().fuse();
                while let Ok(watch_req) = watch_req_receiver.recv().await {
                    let processing_key = watch_req.get_key();
                    futures::select! {
                        res = client_req_sender.send((watch_req.clone().into(), WriteFlags::default())).fuse() => {
                            if let Err(e) = res {
                                warn!("Fail to send watch request, the error is: {}, clean cache", e);
                                this_clone.clean_cache().await;
                                continue;
                            }
                        },
                        _ = shutdown_rx => return
                    };
                    // Wait until etcd server returns watch id.
                    let watch_id = match watch_id_receiver.recv().await {
                        Err(e) => {
                            warn!("Fail to receive watch id from channel, the error is {}, clean cache", e);
                            this_clone.clean_cache().await;
                            continue;
                        }
                        Ok(id) => id,
                    };
                    // Watch request can only be create or cancel.
                    if watch_req.is_create() {
                        cache_clone.insert_watch_id(processing_key.clone(), watch_id);
                    } else {
                        cache_clone.delete_watch_id(&processing_key);
                    }
                }
            })
            .detach();
            // Task that handle the watch responses from Etcd server.
            smol::spawn(async move {
                let mut shutdown_rx = shutdown_response.into_future().fuse();
                loop {
                    let watch_resp = futures::select! {
                        resp_opt = client_resp_receiver.next().fuse() => {
                            if let Some(resp) = resp_opt {
                                resp
                            } else {
                                break;
                            }
                        },
                        _ = shutdown_rx => return
                    };
                    match watch_resp {
                        Ok(resp) => {
                            // TODO: Check if need to spawn new task here.
                            if resp.get_created() || resp.get_canceled() {
                                if let Err(e) = watch_id_sender.send(resp.get_watch_id()).await {
                                    warn!("Fail to send watch id, the error is {}, clean cache", e);
                                    this_clone2.clean_cache().await;
                                    continue;
                                }
                            } else {
                                let events = resp.get_events().to_vec();
                                for event in events {
                                    if event.get_field_type() == Event_EventType::PUT {
                                        if let Some(valid_value) = cache_inner
                                            .search(event.get_kv().get_key().to_vec())
                                            .await
                                        {
                                            // Only update the cache if event's version is larger than
                                            // existing value's version
                                            if valid_value.version < event.get_kv().get_version() {
                                                cache_inner
                                                    .insert(
                                                        event.get_kv().get_key().to_vec(),
                                                        event.get_kv().clone(),
                                                    )
                                                    .await;
                                            }
                                        }
                                    } else {
                                        cache_inner
                                            .delete(event.get_kv().get_key().to_vec(), true)
                                            .await;
                                    }
                                }
                            }
                        }
                        Err(e) => {
                            warn!(
                                "Watch response contains error, the error is: {}, clean cache",
                                e
                            );
                            this_clone2.clean_cache().await;
                            continue;
                        }
                    }
                }
            })
            .detach();
        };
        this
    }
fn new ( client : KvClient , watch_client : & WatchClient , cache_size : usize , cache_enable : bool , ) -> Arc < Self > { let etcd_cache_size = if cache_size == 0 { ETCD_CACHE_DEFAULT_SIZE } else { cache_size } ; let ( watch_req_sender , watch_req_receiver ) = unbounded :: < EtcdWatchRequest > ( ) ; let ( watch_id_sender , watch_id_receiver ) = bounded :: < i64 > ( 1 ) ; let ( shutdown_tx , shutdown_rx ) = unbounded ( ) ; let shutdown_response = shutdown_rx . clone ( ) ; let cache = Cache :: new ( etcd_cache_size , watch_req_sender . clone ( ) ) ; let ( mut client_req_sender , mut client_resp_receiver ) = watch_client . watch ( ) . unwrap_or_else ( | e | panic ! ( " " e ) ) ; let cache_clone = cache . clone ( ) ; let cache_inner = cache . clone ( ) ; let this = Arc :: new ( Self { client , cache , cache_enable , watch_sender : watch_req_sender , cache_availability : RwLock :: new ( ( ) ) , shutdown : shutdown_tx , } ) ; let this_clone = Arc :: < Self > :: clone ( & this ) ; let this_clone2 = Arc :: < Self > :: clone ( & this ) ; if cache_enable { // Task that handles all the pending watch requests. smol :: spawn ( async move { let mut shutdown_rx = shutdown_rx . into_future ( ) . fuse ( ) ; while let Ok ( watch_req ) = watch_req_receiver . recv ( ) . await { let processing_key = watch_req . get_key ( ) ; futures :: select ! { res client_req_sender send ( ( watch_req clone ( ) into ( ) WriteFlags default ( ) ) ) fuse ( ) { if let Err ( e ) res { warn ( " " e ) this_clone clean_cache ( ) await continue } } shutdown_rx return } ; // Wait until etcd server returns watch id. let watch_id = match watch_id_receiver . recv ( ) . await { Err ( e ) => { warn ! ( " " e ) ; this_clone . clean_cache ( ) . await ; continue ; } Ok ( id ) => id , } ; // Watch request can only be create or cancel. if watch_req . is_create ( ) { cache_clone . insert_watch_id ( processing_key . clone ( ) , watch_id ) ; } else { cache_clone . delete_watch_id ( & processing_key ) ; } } } ) . detach ( ) ; // Task that handle the watch responses from Etcd server. smol :: spawn ( async move { let mut shutdown_rx = shutdown_response . into_future ( ) . fuse ( ) ; loop { let watch_resp = futures :: select ! { resp_opt client_resp_receiver next ( ) fuse ( ) { if let Some ( resp ) resp_opt { resp } else { break } } shutdown_rx return } ; match watch_resp { Ok ( resp ) => { // TODO: Check if need to spawn new task here. if resp . get_created ( ) || resp . get_canceled ( ) { if let Err ( e ) = watch_id_sender . send ( resp . get_watch_id ( ) ) . await { warn ! ( " " e ) ; this_clone2 . clean_cache ( ) . await ; continue ; } } else { let events = resp . get_events ( ) . to_vec ( ) ; for event in events { if event . get_field_type ( ) == Event_EventType :: PUT { if let Some ( valid_value ) = cache_inner . search ( event . get_kv ( ) . get_key ( ) . to_vec ( ) ) . await { // Only update the cache if event's version is larger than // existing value's version if valid_value . version < event . get_kv ( ) . get_version ( ) { cache_inner . insert ( event . get_kv ( ) . get_key ( ) . to_vec ( ) , event . get_kv ( ) . clone ( ) , ) . await ; } } } else { cache_inner . delete ( event . get_kv ( ) . get_key ( ) . to_vec ( ) , true ) . await ; } } } } Err ( e ) => { warn ! ( " " e ) ; this_clone2 . clean_cache ( ) . await ; continue ; } } } } ) . detach ( ) ; } ; this }
make no sense to split new()
make no sense to split new ()
client
KvClient
watch_client
&WatchClient
cache_size
usize
cache_enable
bool