risingwave_meta/barrier/
notifier.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use tokio::sync::oneshot;
16
17use crate::{MetaError, MetaResult};
18
19/// Used for notifying the status of a scheduled command/barrier.
20#[derive(Debug, Default)]
21pub(crate) struct Notifier {
22    /// Get notified when scheduled barrier has started to be handled.
23    pub started: Option<oneshot::Sender<MetaResult<()>>>,
24
25    /// Get notified when scheduled barrier is collected or failed.
26    pub collected: Option<oneshot::Sender<MetaResult<()>>>,
27}
28
29impl Notifier {
30    /// Notify when we have injected a barrier to compute nodes.
31    pub fn notify_started(&mut self) {
32        if let Some(tx) = self.started.take() {
33            tx.send(Ok(())).ok();
34        }
35    }
36
37    pub fn notify_start_failed(self, err: MetaError) {
38        if let Some(tx) = self.started {
39            tx.send(Err(err)).ok();
40        }
41    }
42
43    /// Notify when we have collected a barrier from all actors.
44    pub fn notify_collected(self) {
45        if let Some(tx) = self.collected {
46            tx.send(Ok(())).ok();
47        }
48    }
49
50    /// Notify when we failed to collect a barrier. This function consumes `self`.
51    pub fn notify_collection_failed(self, err: MetaError) {
52        if let Some(tx) = self.collected {
53            tx.send(Err(err)).ok();
54        }
55    }
56}
risingwave_meta/barrier/notifier.rs

risingwave_meta/barrier/
notifier.rs
