test(channel): harden Lark WS heartbeat activity handling
This commit is contained in:
parent
3108ffe3e7
commit
adc998429e
1 changed files with 32 additions and 5 deletions
|
|
@ -127,6 +127,12 @@ struct LarkMessage {
|
||||||
/// If no binary frame (pong or event) is received within this window, reconnect.
|
/// If no binary frame (pong or event) is received within this window, reconnect.
|
||||||
const WS_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(300);
|
const WS_HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(300);
|
||||||
|
|
||||||
|
/// Returns true when the WebSocket frame indicates live traffic that should
|
||||||
|
/// refresh the heartbeat watchdog.
|
||||||
|
fn should_refresh_last_recv(msg: &WsMsg) -> bool {
|
||||||
|
matches!(msg, WsMsg::Binary(_) | WsMsg::Ping(_) | WsMsg::Pong(_))
|
||||||
|
}
|
||||||
|
|
||||||
/// Lark/Feishu channel.
|
/// Lark/Feishu channel.
|
||||||
///
|
///
|
||||||
/// Supports two receive modes (configured via `receive_mode` in config):
|
/// Supports two receive modes (configured via `receive_mode` in config):
|
||||||
|
|
@ -321,12 +327,20 @@ impl LarkChannel {
|
||||||
|
|
||||||
msg = read.next() => {
|
msg = read.next() => {
|
||||||
let raw = match msg {
|
let raw = match msg {
|
||||||
Some(Ok(WsMsg::Binary(b))) => { last_recv = Instant::now(); b }
|
Some(Ok(ws_msg)) => {
|
||||||
Some(Ok(WsMsg::Ping(d))) => { last_recv = Instant::now(); let _ = write.send(WsMsg::Pong(d)).await; continue; }
|
if should_refresh_last_recv(&ws_msg) {
|
||||||
Some(Ok(WsMsg::Pong(_))) => { last_recv = Instant::now(); continue; }
|
last_recv = Instant::now();
|
||||||
Some(Ok(WsMsg::Close(_))) | None => { tracing::info!("Lark: WS closed — reconnecting"); break; }
|
}
|
||||||
|
match ws_msg {
|
||||||
|
WsMsg::Binary(b) => b,
|
||||||
|
WsMsg::Ping(d) => { let _ = write.send(WsMsg::Pong(d)).await; continue; }
|
||||||
|
WsMsg::Pong(_) => continue,
|
||||||
|
WsMsg::Close(_) => { tracing::info!("Lark: WS closed — reconnecting"); break; }
|
||||||
|
_ => continue,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => { tracing::info!("Lark: WS closed — reconnecting"); break; }
|
||||||
Some(Err(e)) => { tracing::error!("Lark: WS read error: {e}"); break; }
|
Some(Err(e)) => { tracing::error!("Lark: WS read error: {e}"); break; }
|
||||||
_ => continue,
|
|
||||||
};
|
};
|
||||||
|
|
||||||
let frame = match PbFrame::decode(&raw[..]) {
|
let frame = match PbFrame::decode(&raw[..]) {
|
||||||
|
|
@ -899,6 +913,19 @@ mod tests {
|
||||||
assert_eq!(ch.name(), "lark");
|
assert_eq!(ch.name(), "lark");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn lark_ws_activity_refreshes_heartbeat_watchdog() {
|
||||||
|
assert!(should_refresh_last_recv(&WsMsg::Binary(vec![1, 2, 3])));
|
||||||
|
assert!(should_refresh_last_recv(&WsMsg::Ping(vec![9, 9])));
|
||||||
|
assert!(should_refresh_last_recv(&WsMsg::Pong(vec![8, 8])));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn lark_ws_non_activity_frames_do_not_refresh_heartbeat_watchdog() {
|
||||||
|
assert!(!should_refresh_last_recv(&WsMsg::Text("hello".into())));
|
||||||
|
assert!(!should_refresh_last_recv(&WsMsg::Close(None)));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn lark_user_allowed_exact() {
|
fn lark_user_allowed_exact() {
|
||||||
let ch = make_channel();
|
let ch = make_channel();
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue