chk
/
AppFlowy-Cloud
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore: remove workspace and collab listener (#893)

This commit is contained in:
Nathan.fooo 2024-10-17 17:43:00 +08:00 committed by GitHub
parent 3623d9f296
commit ffee47d0f7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 13 additions and 65 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
services/appflowy-collaborate/src/application.rs
View File

@ -3,7 +3,7 @@ use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use access_control::casbin::collab::{CollabAccessControlImpl, RealtimeCollabAccessControlImpl}; use access_control::casbin::collab::{CollabAccessControlImpl, RealtimeCollabAccessControlImpl};
use access_control::casbin::notification::spawn_listen_on_workspace_member_change;
use access_control::casbin::workspace::WorkspaceAccessControlImpl; use access_control::casbin::workspace::WorkspaceAccessControlImpl;
use actix::Supervisor; use actix::Supervisor;
use actix_web::dev::Server; use actix_web::dev::Server;
@ -23,7 +23,6 @@ use appflowy_ai_client::client::AppFlowyAIClient;
use crate::api::{collab_scope, ws_scope}; use crate::api::{collab_scope, ws_scope};
use crate::collab::notification::spawn_listen_on_collab_member_change;
use crate::collab::storage::CollabStorageImpl; use crate::collab::storage::CollabStorageImpl;
use crate::command::{CLCommandReceiver, CLCommandSender}; use crate::command::{CLCommandReceiver, CLCommandSender};
use crate::config::{Config, DatabaseSetting}; use crate::config::{Config, DatabaseSetting};
@ -119,15 +118,14 @@ pub async fn init_state(config: &Config, rt_cmd_tx: CLCommandSender) -> Result<A
let pg_listeners = Arc::new(PgListeners::new(&pg_pool).await?); let pg_listeners = Arc::new(PgListeners::new(&pg_pool).await?);
let access_control = let access_control =
AccessControl::new(pg_pool.clone(), metrics.access_control_metrics.clone()).await?; AccessControl::new(pg_pool.clone(), metrics.access_control_metrics.clone()).await?;
let collab_member_listener = pg_listeners.subscribe_collab_member_change(); // let collab_member_listener = pg_listeners.subscribe_collab_member_change();
let workspace_member_listener = pg_listeners.subscribe_workspace_member_change(); // let workspace_member_listener = pg_listeners.subscribe_workspace_member_change();
// spawn_listen_on_workspace_member_change(workspace_member_listener, access_control.clone());
spawn_listen_on_workspace_member_change(workspace_member_listener, access_control.clone()); // spawn_listen_on_collab_member_change(
spawn_listen_on_collab_member_change( // pg_pool.clone(),
pg_pool.clone(), // collab_member_listener,
collab_member_listener, // access_control.clone(),
access_control.clone(), // );
);
let collab_access_control = CollabAccessControlImpl::new(access_control.clone()); let collab_access_control = CollabAccessControlImpl::new(access_control.clone());
let workspace_access_control = WorkspaceAccessControlImpl::new(access_control.clone()); let workspace_access_control = WorkspaceAccessControlImpl::new(access_control.clone());

32
services/appflowy-collaborate/src/pg_listener.rs
View File

@ -1,42 +1,16 @@
use crate::collab::notification::CollabMemberNotification;
use access_control::casbin::notification::WorkspaceMemberNotification;
use anyhow::Error; use anyhow::Error;
use database::listener::PostgresDBListener; use database::listener::PostgresDBListener;
use database::pg_row::AFUserNotification; use database::pg_row::AFUserNotification;
use sqlx::PgPool; use sqlx::PgPool;
use tokio::sync::broadcast;
pub struct PgListeners { pub struct PgListeners {
user_listener: UserListener, user_listener: UserListener,
workspace_member_listener: WorkspaceMemberListener,
collab_member_listener: CollabMemberListener,
} }
impl PgListeners { impl PgListeners {
pub async fn new(pg_pool: &PgPool) -> Result<Self, Error> { pub async fn new(pg_pool: &PgPool) -> Result<Self, Error> {
let user_listener = UserListener::new(pg_pool, "af_user_channel").await?; let user_listener = UserListener::new(pg_pool, "af_user_channel").await?;
Ok(Self { user_listener })
let workspace_member_listener =
WorkspaceMemberListener::new(pg_pool, "af_workspace_member_channel").await?;
let collab_member_listener =
CollabMemberListener::new(pg_pool, "af_collab_member_channel").await?;
Ok(Self {
user_listener,
workspace_member_listener,
collab_member_listener,
})
}
pub fn subscribe_workspace_member_change(
&self,
) -> broadcast::Receiver<WorkspaceMemberNotification> {
self.workspace_member_listener.notify.subscribe()
}
pub fn subscribe_collab_member_change(&self) -> broadcast::Receiver<CollabMemberNotification> {
self.collab_member_listener.notify.subscribe()
} }
pub fn subscribe_user_change(&self, uid: i64) -> tokio::sync::mpsc::Receiver<AFUserNotification> { pub fn subscribe_user_change(&self, uid: i64) -> tokio::sync::mpsc::Receiver<AFUserNotification> {
@ -55,6 +29,6 @@ impl PgListeners {
} }
} }
pub type CollabMemberListener = PostgresDBListener<CollabMemberNotification>; // pub type CollabMemberListener = PostgresDBListener<CollabMemberNotification>;
// pub type WorkspaceMemberListener = PostgresDBListener<WorkspaceMemberNotification>;
pub type UserListener = PostgresDBListener<AFUserNotification>; pub type UserListener = PostgresDBListener<AFUserNotification>;
pub type WorkspaceMemberListener = PostgresDBListener<WorkspaceMemberNotification>;

26
src/biz/pg_listener.rs
View File

@ -4,39 +4,15 @@ use appflowy_collaborate::collab::notification::CollabMemberNotification;
use database::listener::PostgresDBListener; use database::listener::PostgresDBListener;
use database::pg_row::AFUserNotification; use database::pg_row::AFUserNotification;
use sqlx::PgPool; use sqlx::PgPool;
use tokio::sync::broadcast;
pub struct PgListeners { pub struct PgListeners {
user_listener: UserListener, user_listener: UserListener,
workspace_member_listener: WorkspaceMemberListener,
collab_member_listener: CollabMemberListener,
} }
impl PgListeners { impl PgListeners {
pub async fn new(pg_pool: &PgPool) -> Result<Self, Error> { pub async fn new(pg_pool: &PgPool) -> Result<Self, Error> {
let user_listener = UserListener::new(pg_pool, "af_user_channel").await?; let user_listener = UserListener::new(pg_pool, "af_user_channel").await?;
Ok(Self { user_listener })
let workspace_member_listener =
WorkspaceMemberListener::new(pg_pool, "af_workspace_member_channel").await?;
let collab_member_listener =
CollabMemberListener::new(pg_pool, "af_collab_member_channel").await?;
Ok(Self {
user_listener,
workspace_member_listener,
collab_member_listener,
})
}
pub fn subscribe_workspace_member_change(
&self,
) -> broadcast::Receiver<WorkspaceMemberNotification> {
self.workspace_member_listener.notify.subscribe()
}
pub fn subscribe_collab_member_change(&self) -> broadcast::Receiver<CollabMemberNotification> {
self.collab_member_listener.notify.subscribe()
} }
pub fn subscribe_user_change(&self, uid: i64) -> tokio::sync::mpsc::Receiver<AFUserNotification> { pub fn subscribe_user_change(&self, uid: i64) -> tokio::sync::mpsc::Receiver<AFUserNotification> {