refactor: database row and cell notification (#5237)

* refactor: database row and cell notification

* chore: clippy

* chore: fix test
This commit is contained in:
Nathan.fooo
2024-05-02 11:42:33 +08:00
committed by GitHub
parent e1e8747f15
commit 7831d8d4ab
19 changed files with 305 additions and 233 deletions

View File

@ -13,6 +13,9 @@ protobuf.workspace = true
tracing.workspace = true
bytes.workspace = true
serde = { workspace = true, features = ["derive"] }
dashmap = "5.5"
tokio-util = "0.7"
tokio = { workspace = true, features = ["time"] }
flowy-derive.workspace = true
lib-dispatch = { workspace = true }

View File

@ -0,0 +1,94 @@
use crate::entities::SubscribeObject;
use crate::NOTIFICATION_SENDER;
use bytes::Bytes;
use lib_dispatch::prelude::ToBytes;
pub struct NotificationBuilder {
/// This identifier is used to uniquely distinguish each notification. For instance, if the
/// notification relates to a folder's view, the identifier could be the view's ID. The frontend
/// uses this ID to link the notification with the relevant observable entity.
id: String,
payload: Option<Bytes>,
error: Option<Bytes>,
source: String,
ty: i32,
}
impl NotificationBuilder {
pub fn new<T: Into<i32>>(id: &str, ty: T, source: &str) -> Self {
Self {
id: id.to_owned(),
ty: ty.into(),
payload: None,
error: None,
source: source.to_owned(),
}
}
pub fn payload<T>(mut self, payload: T) -> Self
where
T: ToBytes,
{
match payload.into_bytes() {
Ok(bytes) => self.payload = Some(bytes),
Err(e) => {
tracing::error!("Set observable payload failed: {:?}", e);
},
}
self
}
pub fn error<T>(mut self, error: T) -> Self
where
T: ToBytes,
{
match error.into_bytes() {
Ok(bytes) => self.error = Some(bytes),
Err(e) => {
tracing::error!("Set observable error failed: {:?}", e);
},
}
self
}
pub fn build(self) -> SubscribeObject {
let payload = self.payload.map(|bytes| bytes.to_vec());
let error = self.error.map(|bytes| bytes.to_vec());
SubscribeObject {
source: self.source,
ty: self.ty,
id: self.id,
payload,
error,
}
}
pub fn send(self) {
let payload = self.payload.map(|bytes| bytes.to_vec());
let error = self.error.map(|bytes| bytes.to_vec());
let subject = SubscribeObject {
source: self.source,
ty: self.ty,
id: self.id,
payload,
error,
};
send_subject(subject);
}
}
#[inline]
pub fn send_subject(subject: SubscribeObject) {
match NOTIFICATION_SENDER.read() {
Ok(read_guard) => read_guard.iter().for_each(|sender| {
if let Err(e) = sender.send_subject(subject.clone()) {
tracing::error!("Post notification failed: {}", e);
}
}),
Err(err) => {
tracing::error!("Read notification sender failed: {}", err);
},
}
}

View File

@ -0,0 +1,54 @@
use crate::entities::SubscribeObject;
use crate::{send_subject, NotificationBuilder};
use dashmap::mapref::entry::Entry;
use dashmap::DashMap;
use lib_dispatch::prelude::ToBytes;
use tokio_util::sync::CancellationToken;
pub struct DebounceNotificationSender {
debounce_in_millis: u64,
cancel_token_by_subject: DashMap<String, CancellationToken>,
}
impl DebounceNotificationSender {
pub fn new(debounce_in_millis: u64) -> Self {
Self {
debounce_in_millis,
cancel_token_by_subject: DashMap::new(),
}
}
pub fn send<T: Into<i32>, P: ToBytes>(&self, id: &str, ty: T, source: &str, payload: P) {
let subject = NotificationBuilder::new(id, ty, source)
.payload(payload)
.build();
self.send_subject(subject);
}
pub fn send_subject(&self, subject: SubscribeObject) {
let subject_key = format!("{}-{}-{}", subject.source, subject.id, subject.ty);
// remove the old cancel token and call cancel to stop the old task
if let Entry::Occupied(entry) = self.cancel_token_by_subject.entry(subject_key.clone()) {
let cancel_token = entry.get();
cancel_token.cancel();
entry.remove();
}
// insert a new cancel token
let cancel_token = CancellationToken::new();
self
.cancel_token_by_subject
.insert(subject_key.clone(), cancel_token.clone());
let debounce_in_millis = self.debounce_in_millis;
tokio::spawn(async move {
if debounce_in_millis > 0 {
tokio::time::sleep(std::time::Duration::from_millis(debounce_in_millis)).await;
}
if cancel_token.is_cancelled() {
return;
}
send_subject(subject);
});
}
}

View File

@ -1,11 +1,12 @@
use std::sync::RwLock;
use bytes::Bytes;
use lazy_static::lazy_static;
use lib_dispatch::prelude::ToBytes;
use crate::entities::SubscribeObject;
use lazy_static::lazy_static;
mod builder;
pub use builder::*;
mod debounce;
pub use debounce::*;
pub mod entities;
mod protobuf;
@ -36,73 +37,3 @@ pub fn unregister_all_notification_sender() {
pub trait NotificationSender: Send + Sync + 'static {
fn send_subject(&self, subject: SubscribeObject) -> Result<(), String>;
}
pub struct NotificationBuilder {
id: String,
payload: Option<Bytes>,
error: Option<Bytes>,
source: String,
ty: i32,
}
impl NotificationBuilder {
pub fn new<T: Into<i32>>(id: &str, ty: T, source: &str) -> Self {
Self {
id: id.to_owned(),
ty: ty.into(),
payload: None,
error: None,
source: source.to_owned(),
}
}
pub fn payload<T>(mut self, payload: T) -> Self
where
T: ToBytes,
{
match payload.into_bytes() {
Ok(bytes) => self.payload = Some(bytes),
Err(e) => {
tracing::error!("Set observable payload failed: {:?}", e);
},
}
self
}
pub fn error<T>(mut self, error: T) -> Self
where
T: ToBytes,
{
match error.into_bytes() {
Ok(bytes) => self.error = Some(bytes),
Err(e) => {
tracing::error!("Set observable error failed: {:?}", e);
},
}
self
}
pub fn send(self) {
let payload = self.payload.map(|bytes| bytes.to_vec());
let error = self.error.map(|bytes| bytes.to_vec());
let subject = SubscribeObject {
source: self.source,
ty: self.ty,
id: self.id,
payload,
error,
};
match NOTIFICATION_SENDER.read() {
Ok(read_guard) => read_guard.iter().for_each(|sender| {
if let Err(e) = sender.send_subject(subject.clone()) {
tracing::error!("Post notification failed: {}", e);
}
}),
Err(err) => {
tracing::error!("Read notification sender failed: {}", err);
},
}
}
}