chore: Bump ver: v0.1.1

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "watcher"
 description = "subscribe data changes"
-version = "0.1.0"
+version = "0.1.1"
 authors = ["Databend Authors <opensource@datafuselabs.com>"]
 license = "Apache-2.0"
 edition = "2021"

src/desc.rs

@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::Bound;
+use std::fmt;
+
 use crate::event_filter::EventFilter;
 use crate::id::WatcherId;
 use crate::type_config::TypeConfig;
@@ -49,3 +52,74 @@ where C: TypeConfig
         }
     }
 }
+
+impl<C> fmt::Display for WatchDesc<C>
+where
+    C: TypeConfig,
+    C::Key: fmt::Display,
+{
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "(id:{} {} ", self.watcher_id, self.interested)?;
+
+        match &self.key_range.0 {
+            Bound::Included(v) => {
+                write!(f, "[{}", v)?;
+            }
+            Bound::Excluded(v) => {
+                write!(f, "({}", v)?;
+            }
+            Bound::Unbounded => {
+                write!(f, "(-∞")?;
+            }
+        }
+
+        match &self.key_range.1 {
+            Bound::Included(v) => {
+                write!(f, ", {}]", v)?;
+            }
+            Bound::Excluded(v) => {
+                write!(f, ", {})", v)?;
+            }
+            Bound::Unbounded => {
+                write!(f, ", +∞)")?;
+            }
+        }
+
+        write!(f, ")")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::ops::Bound;
+
+    use super::*;
+    use crate::testing::UTTypes;
+
+    #[test]
+    fn test_watch_desc_display() {
+        let desc = WatchDesc::<UTTypes>::new(
+            1,
+            EventFilter::all(),
+            (
+                Bound::Included("a".to_string()),
+                Bound::Excluded("z".to_string()),
+            ),
+        );
+        assert_eq!(format!("{}", desc), r#"(id:1 update|delete [a, z))"#);
+
+        let desc = WatchDesc::<UTTypes>::new(
+            1,
+            EventFilter::delete(),
+            (Bound::Excluded("a".to_string()), Bound::Unbounded),
+        );
+        assert_eq!(format!("{}", desc), r#"(id:1 delete (a, +∞))"#);
+
+        let desc = WatchDesc::<UTTypes>::new(
+            1,
+            EventFilter::update(),
+            (Bound::Unbounded, Bound::Included("a".to_string())),
+        );
+        assert_eq!(format!("{}", desc), r#"(id:1 update (-∞, a])"#);
+    }
+}

src/dispatch/dispatcher.rs

@@ -14,7 +14,9 @@
 
 use std::collections::BTreeSet;
 use std::sync::Arc;
+use std::sync::Weak;
 
+use log::debug;
 use log::info;
 use log::warn;
 use span_map::SpanMap;
@@ -47,6 +49,14 @@ where C: TypeConfig
     current_watcher_id: WatcherId,
 }
 
+impl<C> Drop for Dispatcher<C>
+where C: TypeConfig
+{
+    fn drop(&mut self) {
+        debug!("watch-event-Dispatcher is dropped");
+    }
+}
+
 impl<C> Dispatcher<C>
 where C: TypeConfig
 {
@@ -83,7 +93,7 @@ where C: TypeConfig
             }
         }
 
-        info!("watch-event-Dispatcher: all event senders are closed. quit.");
+        info!("watch-event-Dispatcher: all event senders are closed(dropped). quit");
     }
 
     /// Dispatch a kv change event to interested watchers.
@@ -97,7 +107,13 @@ where C: TypeConfig
 
         let mut removed = vec![];
 
+        debug!("watch-event-Dispatcher: dispatch event {:?}", kv_change);
+
         for sender in self.watchers.get(&kv_change.0) {
+            debug!(
+                "watch-event-Dispatcher: dispatch event to watcher {:?}",
+                sender
+            );
             let interested = sender.desc.interested;
 
             if !interested.accepts_event_type(event_type) {
@@ -107,8 +123,8 @@ where C: TypeConfig
             let resp = C::new_response(kv_change.clone());
             if let Err(_err) = sender.send(resp).await {
                 warn!(
-                    "watch-event-Dispatcher: fail to send to watcher {}; close this stream",
-                    sender.desc.watcher_id
+                    "watch-event-Dispatcher: fail to send to watcher {:?}; close this stream",
+                    sender
                 );
                 removed.push(sender.clone());
             };
@@ -124,7 +140,7 @@ where C: TypeConfig
         rng: KeyRange<C>,
         filter: EventFilter,
         tx: mpsc::Sender<WatchResult<C>>,
-    ) -> Arc<WatchStreamSender<C>> {
+    ) -> Weak<WatchStreamSender<C>> {
         info!(
             "watch-event-Dispatcher::add_watcher: range: {:?}, filter: {}",
             rng, filter
@@ -139,7 +155,7 @@ where C: TypeConfig
 
         C::update_watcher_metrics(1);
 
-        stream_sender
+        Arc::downgrade(&stream_sender)
     }
 
     fn new_watch_desc(&mut self, key_range: KeyRange<C>, interested: EventFilter) -> WatchDesc<C> {

src/dispatch/dispatcher_handle.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
+use std::sync::Weak;
 
 use log::error;
 use tokio::sync::mpsc;
@@ -65,7 +66,7 @@ where C: TypeConfig
         key_range: KeyRange<C>,
         filter: EventFilter,
         tx: mpsc::Sender<WatchResult<C>>,
-    ) -> Result<Arc<WatchStreamSender<C>>, &'static str> {
+    ) -> Result<Weak<WatchStreamSender<C>>, &'static str> {
         self.request_blocking(move |dispatcher| dispatcher.add_watcher(key_range, filter, tx))
             .await
             .map_err(|_| "Failed to add watcher; watch-Dispatcher may be closed")

src/lib.rs

@@ -23,6 +23,9 @@ pub mod type_config;
 pub mod util;
 pub mod watch_stream;
 
+#[cfg(test)]
+pub(crate) mod testing;
+
 use std::collections::Bound;
 
 pub use event_filter::EventFilter;

src/testing.rs

@@ -0,0 +1,49 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::future::Future;
+use std::io;
+
+use crate::type_config::KVChange;
+use crate::type_config::TypeConfig;
+
+// Only Debug is actually needed for the test framework
+#[derive(Debug, Copy, Clone)]
+pub(crate) struct UTTypes {}
+
+impl TypeConfig for UTTypes {
+    type Key = String;
+    type Value = String;
+    type Response = (String, Option<String>, Option<String>);
+    type Error = io::Error;
+
+    fn new_response(change: KVChange<Self>) -> Self::Response {
+        change
+    }
+
+    fn data_error(error: io::Error) -> Self::Error {
+        error
+    }
+
+    fn update_watcher_metrics(_delta: i64) {}
+
+    #[allow(clippy::disallowed_methods)]
+    fn spawn<T>(_fut: T)
+    where
+        T: Future + Send + 'static,
+        T::Output: Send + 'static,
+    {
+        todo!()
+    }
+}

src/watch_stream/sender.rs

@@ -16,6 +16,7 @@ use std::cmp::Ordering;
 use std::fmt;
 use std::fmt::Formatter;
 
+use log::debug;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::error::SendError;
 
@@ -28,14 +29,21 @@ use crate::WatchResult;
 /// The stream sender is responsible for sending watch events through the stream
 /// to the client-side watcher. It encapsulates the communication channel between
 /// the server's event source and the client's watch request.
-#[derive(Clone)]
 pub struct WatchStreamSender<C>
 where C: TypeConfig
 {
     pub desc: WatchDesc<C>,
     tx: mpsc::Sender<WatchResult<C>>,
 }
 
+impl<C> Drop for WatchStreamSender<C>
+where C: TypeConfig
+{
+    fn drop(&mut self) {
+        debug!("WatchStreamSender({:?}) dropped", self.desc,);
+    }
+}
+
 impl<C> fmt::Debug for WatchStreamSender<C>
 where C: TypeConfig
 {
@@ -44,6 +52,16 @@ where C: TypeConfig
     }
 }
 
+impl<C> fmt::Display for WatchStreamSender<C>
+where
+    C: TypeConfig,
+    C::Key: fmt::Display,
+{
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "WatchStreamSender({})", self.desc)
+    }
+}
+
 impl<C> PartialEq for WatchStreamSender<C>
 where C: TypeConfig
 {

src/watch_stream/stream.rs

@@ -18,6 +18,7 @@ use std::task::Context;
 use std::task::Poll;
 
 use futures::Stream;
+use log::debug;
 use tokio::sync::mpsc::Receiver;
 
 /// A wrapper around [`tokio::sync::mpsc::Receiver`] that implements [`Stream`].
@@ -35,6 +36,7 @@ impl<T> fmt::Debug for WatchStream<T> {
 
 impl<T> Drop for WatchStream<T> {
     fn drop(&mut self) {
+        debug!("WatchStream is dropped");
         let Some(on_drop) = self.on_drop.take() else {
             return;
         };

tests/it/dispatcher.rs

@@ -177,10 +177,16 @@ async fn test_dispatcher_overlapping_ranges() {
     .await;
 
     let watcher1_clone = watcher1.clone();
-    handle.remove_watcher(watcher1_clone).await.unwrap();
+    handle
+        .remove_watcher(watcher1_clone.upgrade().unwrap())
+        .await
+        .unwrap();
 
     let watcher2_clone = watcher2.clone();
-    handle.remove_watcher(watcher2_clone).await.unwrap();
+    handle
+        .remove_watcher(watcher2_clone.upgrade().unwrap())
+        .await
+        .unwrap();
 }
 
 #[tokio::test]
@@ -229,8 +235,12 @@ async fn test_dispatcher_basic_functionality() {
     expect_no_event(&mut rx2, "w2 should not receive (filter excludes DELETE)").await;
 
     // Remove first watcher
+    let watcher1_strong = watcher1.upgrade().unwrap();
     let watcher1_clone = watcher1.clone();
-    handle.remove_watcher(watcher1_clone).await.unwrap();
+    handle
+        .remove_watcher(watcher1_clone.upgrade().unwrap())
+        .await
+        .unwrap();
 
     tokio::time::sleep(REMOVAL_DELAY).await;
     drop(rx1);
@@ -241,11 +251,17 @@ async fn test_dispatcher_basic_functionality() {
     // Verify watcher status
     let senders = all_senders(&handle).await;
     assert_eq!(senders.len(), 1);
-    assert!(!senders.contains(&watcher1), "w1 should be removed");
-    assert!(senders.contains(&watcher2), "w2 should still exist");
+    assert!(!senders.contains(&watcher1_strong), "w1 should be removed");
+    assert!(
+        senders.contains(&watcher2.upgrade().unwrap()),
+        "w2 should still exist"
+    );
 
     let watcher2_clone = watcher2.clone();
-    handle.remove_watcher(watcher2_clone).await.unwrap();
+    handle
+        .remove_watcher(watcher2_clone.upgrade().unwrap())
+        .await
+        .unwrap();
 }
 
 #[tokio::test]
@@ -272,13 +288,16 @@ async fn test_dispatcher_watch_senders() {
     let senders = all_senders(&handle).await;
     assert_eq!(senders.len(), 3);
     for watcher in &watchers {
-        assert!(senders.contains(watcher));
+        assert!(senders.contains(&watcher.upgrade().unwrap()));
     }
 
     // Clean up
     for watcher in watchers.clone() {
         let watcher_clone = watcher.clone();
-        handle.remove_watcher(watcher_clone).await.unwrap();
+        handle
+            .remove_watcher(watcher_clone.upgrade().unwrap())
+            .await
+            .unwrap();
     }
 
     let senders = all_senders(&handle).await;

tests/it/main.rs

@@ -12,4 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![allow(clippy::mutable_key_type)]
+
 mod dispatcher;