chore: drop group with maximum timeout (#747)
This commit is contained in:
parent
248834651b
commit
364d31f825
|
|
@ -8,7 +8,7 @@ use tracing::{instrument, trace, warn};
|
||||||
|
|
||||||
use access_control::collab::RealtimeAccessControl;
|
use access_control::collab::RealtimeAccessControl;
|
||||||
use collab_rt_entity::user::RealtimeUser;
|
use collab_rt_entity::user::RealtimeUser;
|
||||||
use collab_rt_entity::{ClientCollabMessage, ServerCollabMessage, SinkMessage};
|
use collab_rt_entity::{AckCode, ClientCollabMessage, ServerCollabMessage, SinkMessage};
|
||||||
use collab_rt_entity::{CollabAck, RealtimeMessage};
|
use collab_rt_entity::{CollabAck, RealtimeMessage};
|
||||||
use database::collab::CollabStorage;
|
use database::collab::CollabStorage;
|
||||||
|
|
||||||
|
|
@ -155,7 +155,11 @@ where
|
||||||
let origin = first_message.origin().clone();
|
let origin = first_message.origin().clone();
|
||||||
let msg_id = first_message.msg_id();
|
let msg_id = first_message.msg_id();
|
||||||
let object_id = first_message.object_id().to_string();
|
let object_id = first_message.object_id().to_string();
|
||||||
let ack = CollabAck::new(origin, object_id, msg_id, 0);
|
|
||||||
|
// when the group with given id is not found and the the first message is not init sync.
|
||||||
|
// Return AckCode::CannotApplyUpdate to the client and then client will send the init sync message.
|
||||||
|
let ack =
|
||||||
|
CollabAck::new(origin, object_id, msg_id, 0).with_code(AckCode::CannotApplyUpdate);
|
||||||
entry
|
entry
|
||||||
.value()
|
.value()
|
||||||
.send_message(ServerCollabMessage::ClientAck(ack).into())
|
.send_message(ServerCollabMessage::ClientAck(ack).into())
|
||||||
|
|
|
||||||
|
|
@ -11,7 +11,7 @@ use collab_entity::CollabType;
|
||||||
use dashmap::DashMap;
|
use dashmap::DashMap;
|
||||||
use futures_util::{SinkExt, StreamExt};
|
use futures_util::{SinkExt, StreamExt};
|
||||||
use tokio::sync::{mpsc, RwLock};
|
use tokio::sync::{mpsc, RwLock};
|
||||||
use tracing::{error, event, trace};
|
use tracing::{error, event, info, trace};
|
||||||
use yrs::updates::decoder::Decode;
|
use yrs::updates::decoder::Decode;
|
||||||
use yrs::updates::encoder::Encode;
|
use yrs::updates::encoder::Encode;
|
||||||
use yrs::Update;
|
use yrs::Update;
|
||||||
|
|
@ -200,7 +200,26 @@ impl CollabGroup {
|
||||||
modified_at.elapsed().as_secs() > 60 * 60
|
modified_at.elapsed().as_secs() > 60 * 60
|
||||||
} else {
|
} else {
|
||||||
let elapsed_secs = modified_at.elapsed().as_secs();
|
let elapsed_secs = modified_at.elapsed().as_secs();
|
||||||
elapsed_secs > self.timeout_secs() && self.subscribers.is_empty()
|
if elapsed_secs > self.timeout_secs() {
|
||||||
|
// Mark the group as inactive if it has been inactive for more than 3 hours, regardless of the number of subscribers.
|
||||||
|
// Otherwise, return `true` only if there are no subscribers remaining in the group.
|
||||||
|
// If a client modifies a group that has already been marked as inactive (removed),
|
||||||
|
// the client will automatically send an initialization sync to reinitialize the group.
|
||||||
|
const MAXIMUM_SECS: u64 = 3 * 60 * 60;
|
||||||
|
if elapsed_secs > MAXIMUM_SECS {
|
||||||
|
info!(
|
||||||
|
"Group:{} is inactive for {} seconds, subscribers: {}",
|
||||||
|
self.object_id,
|
||||||
|
modified_at.elapsed().as_secs(),
|
||||||
|
self.subscribers.len()
|
||||||
|
);
|
||||||
|
true
|
||||||
|
} else {
|
||||||
|
self.subscribers.is_empty()
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
false
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue