From 3508262d1d9b82dbba235502b676b4fd07f67b65 Mon Sep 17 00:00:00 2001
From: "Nathan.fooo" <86001920+appflowy@users.noreply.github.com>
Date: Tue, 7 May 2024 22:15:53 +0800
Subject: [PATCH] chore: using env to control mailbox size (#532)

* chore: using env to control mailbox size

* chore: update logs
---
 deploy.env                                    |  1 +
 dev.env                                       |  1 +
 .../src/group/broadcast.rs                    | 10 ++-------
 src/biz/actix_ws/server/rt_actor.rs           | 21 ++++++++++++++++---
 4 files changed, 22 insertions(+), 11 deletions(-)

diff --git a/deploy.env b/deploy.env
index 8b0a2e13..a1c399d6 100644
--- a/deploy.env
+++ b/deploy.env
@@ -7,6 +7,7 @@ APPFLOWY_GOTRUE_BASE_URL=http://gotrue:9999
 ## URL that connects to the postgres docker container
 APPFLOWY_DATABASE_URL=postgres://postgres:password@postgres:5432/postgres
 APPFLOWY_ACCESS_CONTROL=true
+APPFLOWY_WEBSOCKET_MAILBOX_SIZE=6000
 
 # admin frontend
 ## URL that connects to redis docker container
diff --git a/dev.env b/dev.env
index e18e7a02..4c91a531 100644
--- a/dev.env
+++ b/dev.env
@@ -2,6 +2,7 @@
 APPFLOWY_GOTRUE_BASE_URL=http://localhost:9999
 APPFLOWY_DATABASE_URL=postgres://postgres:password@localhost:5432/postgres
 APPFLOWY_ACCESS_CONTROL=true
+APPFLOWY_WEBSOCKET_MAILBOX_SIZE=6000
 
 # This file is used to set the environment variables for local development
 # Copy this file to .env and change the values as needed
diff --git a/services/appflowy-collaborate/src/group/broadcast.rs b/services/appflowy-collaborate/src/group/broadcast.rs
index 3eae25ec..8723e73a 100644
--- a/services/appflowy-collaborate/src/group/broadcast.rs
+++ b/services/appflowy-collaborate/src/group/broadcast.rs
@@ -536,18 +536,12 @@ impl Subscription {
   pub async fn stop(&mut self) {
     if let Some(sink_stop_tx) = self.sink_stop_tx.take() {
       if let Err(err) = sink_stop_tx.send(()).await {
-        warn!(
-          "fail to stop sink:{}, the stream might be already stop",
-          err
-        );
+        warn!("the sink might be already stop, error: {}", err);
       }
     }
     if let Some(stream_stop_tx) = self.stream_stop_tx.take() {
       if let Err(err) = stream_stop_tx.send(()).await {
-        warn!(
-          "fail to stop stream:{}, the stream might be already stop",
-          err
-        );
+        warn!("the stream might be already stop, error: {}", err);
       }
     }
   }
diff --git a/src/biz/actix_ws/server/rt_actor.rs b/src/biz/actix_ws/server/rt_actor.rs
index 4471aae0..cb8b74ed 100644
--- a/src/biz/actix_ws/server/rt_actor.rs
+++ b/src/biz/actix_ws/server/rt_actor.rs
@@ -6,7 +6,7 @@ use appflowy_collaborate::{CollaborationServer, RealtimeAccessControl};
 use collab_rt_entity::user::UserDevice;
 use database::collab::CollabStorage;
 use std::ops::Deref;
-use tracing::{error, warn};
+use tracing::{error, info, warn};
 
 #[derive(Clone)]
 pub struct RealtimeServerActor<S, AC>(pub CollaborationServer<S, AC>);
@@ -34,7 +34,12 @@ where
   type Context = Context<Self>;
 
   fn started(&mut self, ctx: &mut Self::Context) {
-    ctx.set_mailbox_capacity(6000);
+    let mail_box_size = mail_box_size();
+    info!(
+      "realtime server started with mailbox size: {}",
+      mail_box_size
+    );
+    ctx.set_mailbox_capacity(mail_box_size);
   }
 }
 impl<S, AC> actix::Supervised for RealtimeServerActor<S, AC>
@@ -44,7 +49,17 @@ where
 {
   fn restarting(&mut self, ctx: &mut Context<RealtimeServerActor<S, AC>>) {
     error!("realtime server is restarting");
-    ctx.set_mailbox_capacity(6000);
+    ctx.set_mailbox_capacity(mail_box_size());
+  }
+}
+
+fn mail_box_size() -> usize {
+  match std::env::var("APPFLOWY_WEBSOCKET_MAILBOX_SIZE") {
+    Ok(value) => value.parse::<usize>().unwrap_or_else(|_| {
+      error!("Error: Invalid mailbox size format, defaulting to 6000");
+      6000
+    }),
+    Err(_) => 6000,
   }
 }