zhenyi

Fork 0

Clone

perf(events): relax idle delivery polling and extend push lease

ZhenYi committed 3 days ago

parent 62f5365 commit 213a71b Browse files

lib/infra/actor/src/push_store.rs +1 −1 View

@@ -113,7 +113,7 @@

113113 ) -> Result<bool, String> {

114114 let field = Self::device(device, namespace);

115115 let mut command = redis::cmd("EVAL");

116 command.arg("local current=redis.call('GET',KEYS[1]); if current and current~=ARGV[1] then return 0 end; redis.call('SET',KEYS[1],ARGV[1],'PX',30000); redis.call('HSETNX',KEYS[2],ARGV[2],'0-0'); return 1")

116 command.arg("local current=redis.call('GET',KEYS[1]); if current and current~=ARGV[1] then return 0 end; redis.call('SET',KEYS[1],ARGV[1],'PX',60000); redis.call('HSETNX',KEYS[2],ARGV[2],'0-0'); return 1")

117117 .arg(2).arg(self.key(user_id, &format!("lease:{field}"))).arg(self.key(user_id,"cursors"))

118118 .arg(owner).arg(field);

119119 self.query::<i64>(&command).await.map(|result| result == 1)

lib/share/events/src/gateway/socket_backend.rs +5 −2 View

@@ -9,6 +9,9 @@

99use crate::gateway::platform::{PlatformClient, PlatformError};

1010use crate::gateway::subscriptions::Subscriptions;

1111

12const IDLE_FALLBACK: Duration = Duration::from_secs(30);

13const STANDBY_WAIT: Duration = Duration::from_secs(2);

14

1215/// socket 通道与事件业务之间的适配层：身份、订阅、投递策略都落在这里。

1316#[derive(Clone)]

1417pub struct RealtimeBackend {

@@ -121,7 +124,7 @@

121124 store.claim(user, &device, &namespace, &owner).await,

122125 Ok(true)

123126 ) {

124 tokio::time::sleep(Duration::from_secs(1)).await;

127 tokio::time::sleep(STANDBY_WAIT).await;

125128 continue;

126129 }

127130 let (cursor, next) = match store.next(user, &device, &namespace).await {

@@ -134,7 +137,7 @@

134137 let Some((delivery, mut payload)) = next else {

135138 tokio::select! {

136139 _ = &mut changed => {}

137 _ = tokio::time::sleep(Duration::from_secs(1)) => {}

140 _ = tokio::time::sleep(IDLE_FALLBACK) => {}

138141 }

139142 continue;

140143 };