feat: watcher subscribes to data source for changes

Author: drmingdrmer

.github/workflows/ci.yml

@@ -0,0 +1,114 @@
+name: ci
+
+on:
+  push:
+  pull_request:
+  schedule: [cron: "40 1 * * 0"]
+
+jobs:
+
+  test-release-build:
+    name: Build
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        include:
+          - toolchain: "nightly"
+
+    steps:
+      - name: Setup | Checkout
+        uses: actions/checkout@v2
+
+
+      - name: Setup | Toolchain
+        uses: actions-rs/toolchain@v1.0.6
+        with:
+          toolchain: "${{ matrix.toolchain }}"
+          override: true
+
+
+      - name: Build | Release Mode
+        uses: actions-rs/cargo@v1
+        with:
+          command: build
+          args: --release
+
+
+  unittest:
+    runs-on: ubuntu-latest
+
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - toolchain: "nightly"
+            features: ""
+
+    steps:
+      - name: Setup | Checkout
+        uses: actions/checkout@v2
+
+
+      - name: Setup | Toolchain
+        uses: actions-rs/toolchain@v1.0.6
+        with:
+          toolchain: "${{ matrix.toolchain }}"
+          override: true
+
+
+      - name: Test
+        uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --features "${{ matrix.features }}"
+        env:
+          RUST_LOG: debug
+          RUST_BACKTRACE: full
+
+
+  lint:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: actions-rs/toolchain@v1.0.6
+        with:
+          toolchain: nightly
+          components: rustfmt, clippy
+
+
+      - name: Check Apache License Header
+        uses: korandoru/hawkeye@v2
+
+
+      - name: Format
+        uses: actions-rs/cargo@v1
+        with:
+          command: fmt
+          args: --all -- --check
+
+
+      - name: clippy
+        shell: bash
+        run: |
+          cargo clippy --no-deps --workspace --all-targets                -- -D warnings
+
+
+      - name: Build-doc
+        uses: actions-rs/cargo@v1
+        with:
+          command: doc
+          args: --all --no-deps
+        env:
+          RUSTDOCFLAGS: "-D warnings"
+
+
+      - shell: bash
+        run: cargo install cargo-audit
+
+
+      - name: Audit dependencies
+        shell: bash
+        # if: "!contains(github.event.head_commit.message, 'skip audit')"
+        run: cargo audit --db ./target/advisory-db

.github/workflows/commit-message-check.yml

@@ -0,0 +1,56 @@
+name: 'commit-message-check'
+on:
+  pull_request:
+    types:
+      - opened
+      - edited
+      - reopened
+      - synchronize
+  push:
+    branches:
+
+jobs:
+  check-commit-message:
+    name: check-subject
+    runs-on: ubuntu-latest
+    steps:
+      - name: check-subject-type
+        uses: gsactions/commit-message-checker@v1
+        with:
+          checkAllCommitMessages: 'true' # optional: this checks all commits associated with a pull request
+          excludeDescription: 'true' # optional: this excludes the description body of a pull request
+          excludeTitle: 'true' # optional: this excludes the title of a pull request
+          accessToken: ${{ secrets.GITHUB_TOKEN }}
+          pattern: '^(DataChange:|Change:|Feature:|Improve:|Perf:|Dep:|Doc:|Test:|CI:|Refactor:|Fix:|Fixdoc:|Fixup:|Merge|BumpVer:|Chore:|fix:|feat:|perf:|refactor:|test:|docs:|deps:|chore:|ci:|Build\(deps\):) .+$'
+          flags: 'gm'
+          error: |
+            Subject line has to contain a commit type, e.g.: "Change: blabla" or a merge commit e.g.: "Merge xxx".
+            Valid types are:
+              DataChange    - Persistent data change
+              Change        - API breaking change
+              Feature       - API compatible new feature
+              feat          - API compatible new feature
+              Improve       - Become better without functional changes
+              Perf          - Performance improvement
+              perf          - Performance improvement
+              Dep           - dependency update
+              deps          - dependency update
+              Doc           - doc update
+              docs          - doc update
+              Test          - test udpate
+              test          - test udpate
+              CI            - CI workflow update
+              ci            - CI workflow update
+              Refactor      - refactor without function change.
+              refactor      - refactor without function change.
+              Fix           - fix bug
+              fix           - fix bug
+              Fixdoc        - fix doc
+              Fixup         - minor change: e.g., fix sth mentioned in a review.
+              BumpVer       - Bump to a new version.
+              Chore         - Nothing important.
+              chore         - Nothing important.
+              Build(deps)   - bot: dependabot.
+
+
+

.gitignore

@@ -14,4 +14,8 @@ target/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+/.idea/
+
+Cargo.lock
+
+.DS_Store

Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "watcher"
+description = "subscribe data changes"
+version = "0.1.0"
+authors = ["Databend Authors <opensource@datafuselabs.com>"]
+license = "Apache-2.0"
+edition = "2021"
+
+[features]
+
+[dependencies]
+futures = "0.3.24"
+log = { version = "0.4.21", features = ["serde", "kv_unstable_std"] }
+span-map = { version = "0.2.0" }
+tokio = { version = "1.35.0", features = ["sync"] }
+tokio-util = { version = "0.7.13" }
+
+[dev-dependencies]
+anyhow = { version = "1.0.65" }
+# Enable other features for testing
+tokio = { version = "1.35.0", features = ["sync", "time", "macros", "rt"] }
+
+#[[example]]
+#name = "basic_usage"
+#path = "examples/basic_usage.rs"

Makefile

@@ -0,0 +1,15 @@
+all: test fmt clippy doc
+
+test:
+	cargo test
+
+fmt:
+	cargo fmt
+
+clippy:
+	cargo clippy
+
+doc:
+	cargo doc
+
+

licenserc.toml

@@ -0,0 +1,38 @@
+baseDir = "."
+
+headerPath = "Apache-2.0.txt"
+
+excludes = [
+    # hidden files
+    ".cargo",
+    ".github",
+    ".dockerignore",
+    ".gitignore",
+    ".gitattributes",
+    ".editorconfig",
+
+    "LICENSE",
+    "Makefile",
+
+    "examples/**",
+
+    # docs and generated files
+    "**/*.md",
+    "**/*.hbs",
+    "**/*.template",
+    "**/*.cue",
+    "**/*.json",
+    "**/*.sql",
+    "**/*.proto",
+    "**/*.yml",
+    "**/*.yaml",
+    "**/*.toml",
+    "**/*.lock",
+    "**/*.yapf",
+    "**/*.test",
+    "**/*.txt",
+]
+
+[properties]
+inceptionYear = 2021
+copyrightOwner = "Datafuse Labs"

rust-toolchain.toml

@@ -0,0 +1,3 @@
+[toolchain]
+channel = "nightly-2025-01-01"
+components = ["rustfmt", "clippy", "rust-src", "miri", "rust-analyzer"]

rustfmt.toml

@@ -0,0 +1,10 @@
+edition = "2021"
+style_edition = "2021"
+reorder_imports = true
+imports_granularity = "Item"
+group_imports = "StdExternalCrate"
+where_single_line = true
+trailing_comma = "Vertical"
+overflow_delimited_expr = true
+format_code_in_doc_comments = true
+normalize_comments = true

src/desc.rs

@@ -0,0 +1,51 @@
+// 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 crate::event_filter::EventFilter;
+use crate::id::WatcherId;
+use crate::type_config::TypeConfig;
+use crate::KeyRange;
+
+/// Descriptor for a watcher that monitors key-value change events.
+///
+/// A `WatchDesc` defines the scope and filtering criteria for a watcher,
+/// specifying which key range to observe and what types of events
+/// (updates, deletes) to receive notifications for.
+#[derive(Clone, Debug)]
+pub struct WatchDesc<C>
+where C: TypeConfig
+{
+    /// Unique identifier for this watcher instance.
+    pub watcher_id: WatcherId,
+
+    /// Event filter that determines which event types (update/delete)
+    /// this watcher should receive.
+    pub interested: EventFilter,
+
+    /// The range of keys this watcher is monitoring.
+    /// Only changes to keys within this range will trigger notifications.
+    pub key_range: KeyRange<C>,
+}
+
+impl<C> WatchDesc<C>
+where C: TypeConfig
+{
+    pub(crate) fn new(id: WatcherId, interested: EventFilter, key_range: KeyRange<C>) -> Self {
+        Self {
+            watcher_id: id,
+            interested,
+            key_range,
+        }
+    }
+}

src/dispatch/command.rs

@@ -0,0 +1,61 @@
+// 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 futures::future::BoxFuture;
+
+use crate::dispatch::Dispatcher;
+use crate::type_config::KVChange;
+use crate::type_config::TypeConfig;
+
+/// A command sent to [`Dispatcher`].
+///
+/// [`Dispatcher`] is a single task that processes all commands serially in the order they are received.
+///
+/// Commands can be categorized into two types:
+/// - Synchronous commands: These commands (like `Change` and `Func`) execute immediately and block
+///   the dispatcher's thread until completion.
+/// - Asynchronous commands: These commands (like `AsyncFunc` and `Future`) return a future that
+///   will be awaited on the dispatcher's thread, allowing other tasks to make progress while waiting.
+///
+/// This command-based architecture ensures thread safety by serializing all operations on the dispatcher.
+#[allow(clippy::type_complexity)]
+pub enum Command<C>
+where C: TypeConfig
+{
+    /// Submit a key-value change event to dispatcher.
+    Change(KVChange<C>),
+
+    /// Execute a function to the [`Dispatcher`].
+    ///
+    /// The function will be called with a mutable reference to the dispatcher,
+    /// allowing it to modify the dispatcher's state or perform operations.
+    /// This is a synchronous operation that blocks until completion.
+    Func {
+        req: Box<dyn FnOnce(&mut Dispatcher<C>) + Send + 'static>,
+    },
+
+    /// Send a function to [`Dispatcher`] to run it asynchronously.
+    ///
+    /// The function will be called with a mutable reference to the dispatcher,
+    /// allowing it to modify the dispatcher's state or perform operations.
+    /// The function returns a future that will be awaited on the dispatcher's thread.
+    AsyncFunc {
+        req: Box<dyn FnOnce(&mut Dispatcher<C>) -> BoxFuture<'static, ()> + Send + 'static>,
+    },
+
+    /// Send a future to [`Dispatcher`] to run it.
+    ///
+    /// The future will be awaited on the dispatcher's thread.
+    Future(BoxFuture<'static, ()>),
+}