harald/zeroclaw
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 3

perf(build): gate Matrix channel for faster iteration

Browse source
This commit is contained in:
Chummy 2026-02-19 21:01:26 +08:00
parent 87dcda638c
commit 05404c6e7a
9 changed files with 135 additions and 60 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/channels/dingtalk.rs
View file

@ -169,7 +169,7 @@ impl Channel for DingTalkChannel {
_ => continue,
};
let frame: serde_json::Value = match serde_json::from_str(&msg) {
let frame: serde_json::Value = match serde_json::from_str(msg.as_ref()) {
Ok(v) => v,
Err(_) => continue,
};
@ -195,7 +195,7 @@ impl Channel for DingTalkChannel {
"data": "",
});
if let Err(e) = write.send(Message::Text(pong.to_string())).await {
if let Err(e) = write.send(Message::Text(pong.to_string().into())).await {
tracing::warn!("DingTalk: failed to send pong: {e}");
break;
}
@ -262,7 +262,7 @@ impl Channel for DingTalkChannel {
"message": "OK",
"data": "",
});
let _ = write.send(Message::Text(ack.to_string())).await;
let _ = write.send(Message::Text(ack.to_string().into())).await;
let channel_msg = ChannelMessage {
id: Uuid::new_v4().to_string(),

10
src/channels/discord.rs
View file

@ -272,7 +272,9 @@ impl Channel for DiscordChannel {
}
}
});
write.send(Message::Text(identify.to_string())).await?;
write
.send(Message::Text(identify.to_string().into()))
.await?;
tracing::info!("Discord: connected and identified");
@ -301,7 +303,7 @@ impl Channel for DiscordChannel {
_ = hb_rx.recv() => {
let d = if sequence >= 0 { json!(sequence) } else { json!(null) };
let hb = json!({"op": 1, "d": d});
if write.send(Message::Text(hb.to_string())).await.is_err() {
if write.send(Message::Text(hb.to_string().into())).await.is_err() {
break;
}
}
@ -312,7 +314,7 @@ impl Channel for DiscordChannel {
_ => continue,
};
let event: serde_json::Value = match serde_json::from_str(&msg) {
let event: serde_json::Value = match serde_json::from_str(msg.as_ref()) {
Ok(e) => e,
Err(_) => continue,
};
@ -329,7 +331,7 @@ impl Channel for DiscordChannel {
1 => {
let d = if sequence >= 0 { json!(sequence) } else { json!(null) };
let hb = json!({"op": 1, "d": d});
if write.send(Message::Text(hb.to_string())).await.is_err() {
if write.send(Message::Text(hb.to_string().into())).await.is_err() {
break;
}
continue;

14
src/channels/lark.rs
View file

@ -288,7 +288,7 @@ impl LarkChannel {
payload: None,
};
if write
.send(WsMsg::Binary(initial_ping.encode_to_vec()))
.send(WsMsg::Binary(initial_ping.encode_to_vec().into()))
.await
.is_err()
{
@ -309,7 +309,7 @@ impl LarkChannel {
headers: vec![PbHeader { key: "type".into(), value: "ping".into() }],
payload: None,
};
if write.send(WsMsg::Binary(ping.encode_to_vec())).await.is_err() {
if write.send(WsMsg::Binary(ping.encode_to_vec().into())).await.is_err() {
tracing::warn!("Lark: ping failed, reconnecting");
break;
}
@ -378,7 +378,7 @@ impl LarkChannel {
let mut ack = frame.clone();
ack.payload = Some(br#"{"code":200,"headers":{},"data":[]}"#.to_vec());
ack.headers.push(PbHeader { key: "biz_rt".into(), value: "0".into() });
let _ = write.send(WsMsg::Binary(ack.encode_to_vec())).await;
let _ = write.send(WsMsg::Binary(ack.encode_to_vec().into())).await;
}
// Fragment reassembly
@ -917,9 +917,11 @@ mod tests {
#[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])));
assert!(should_refresh_last_recv(&WsMsg::Binary(
vec![1, 2, 3].into()
)));
assert!(should_refresh_last_recv(&WsMsg::Ping(vec![9, 9].into())));
assert!(should_refresh_last_recv(&WsMsg::Pong(vec![8, 8].into())));
}
#[test]

28
src/channels/mod.rs
View file

@ -6,6 +6,7 @@ pub mod imessage;
pub mod irc;
pub mod lark;
pub mod linq;
#[cfg(feature = "channel-matrix")]
pub mod matrix;
pub mod mattermost;
pub mod qq;
@ -27,6 +28,7 @@ pub use imessage::IMessageChannel;
pub use irc::IrcChannel;
pub use lark::LarkChannel;
pub use linq::LinqChannel;
#[cfg(feature = "channel-matrix")]
pub use matrix::MatrixChannel;
pub use mattermost::MattermostChannel;
pub use qq::QQChannel;
@ -1389,7 +1391,10 @@ pub async fn handle_command(command: crate::ChannelCommands, config: &Config) ->
("Mattermost", config.channels_config.mattermost.is_some()),
("Webhook", config.channels_config.webhook.is_some()),
("iMessage", config.channels_config.imessage.is_some()),
("Matrix", config.channels_config.matrix.is_some()),
(
"Matrix",
cfg!(feature = "channel-matrix") && config.channels_config.matrix.is_some(),
),
("Signal", config.channels_config.signal.is_some()),
("WhatsApp", config.channels_config.whatsapp.is_some()),
("Linq", config.channels_config.linq.is_some()),
@ -1401,6 +1406,11 @@ pub async fn handle_command(command: crate::ChannelCommands, config: &Config) ->
] {
println!(" {} {name}", if configured { "✅" } else { "❌" });
}
if !cfg!(feature = "channel-matrix") {
println!(
" ℹ️ Matrix channel support is disabled in this build (enable `channel-matrix`)."
);
}
println!("\nTo start channels: zeroclaw channel start");
println!("To check health: zeroclaw channel doctor");
println!("To configure: zeroclaw onboard");
@ -1489,6 +1499,7 @@ pub async fn doctor_channels(config: Config) -> Result<()> {
));
}
#[cfg(feature = "channel-matrix")]
if let Some(ref mx) = config.channels_config.matrix {
channels.push((
"Matrix",
@ -1503,6 +1514,13 @@ pub async fn doctor_channels(config: Config) -> Result<()> {
));
}
#[cfg(not(feature = "channel-matrix"))]
if config.channels_config.matrix.is_some() {
tracing::warn!(
"Matrix channel is configured but this build was compiled without `channel-matrix`; skipping Matrix health check."
);
}
if let Some(ref sig) = config.channels_config.signal {
channels.push((
"Signal",
@ -1864,6 +1882,7 @@ pub async fn start_channels(config: Config) -> Result<()> {
channels.push(Arc::new(IMessageChannel::new(im.allowed_contacts.clone())));
}
#[cfg(feature = "channel-matrix")]
if let Some(ref mx) = config.channels_config.matrix {
channels.push(Arc::new(MatrixChannel::new_with_session_hint(
mx.homeserver.clone(),
@ -1875,6 +1894,13 @@ pub async fn start_channels(config: Config) -> Result<()> {
)));
}
#[cfg(not(feature = "channel-matrix"))]
if config.channels_config.matrix.is_some() {
tracing::warn!(
"Matrix channel is configured but this build was compiled without `channel-matrix`; skipping Matrix runtime startup."
);
}
if let Some(ref sig) = config.channels_config.signal {
channels.push(Arc::new(SignalChannel::new(
sig.http_url.clone(),

18
src/channels/qq.rs
View file

@ -263,7 +263,9 @@ impl Channel for QQChannel {
}
}
});
write.send(Message::Text(identify.to_string())).await?;
write
.send(Message::Text(identify.to_string().into()))
.await?;
tracing::info!("QQ: connected and identified");
@ -287,7 +289,11 @@ impl Channel for QQChannel {
_ = hb_rx.recv() => {
let d = if sequence >= 0 { json!(sequence) } else { json!(null) };
let hb = json!({"op": 1, "d": d});
if write.send(Message::Text(hb.to_string())).await.is_err() {
if write
.send(Message::Text(hb.to_string().into()))
.await
.is_err()
{
break;
}
}
@ -298,7 +304,7 @@ impl Channel for QQChannel {
_ => continue,
};
let event: serde_json::Value = match serde_json::from_str(&msg) {
let event: serde_json::Value = match serde_json::from_str(msg.as_ref()) {
Ok(e) => e,
Err(_) => continue,
};
@ -315,7 +321,11 @@ impl Channel for QQChannel {
1 => {
let d = if sequence >= 0 { json!(sequence) } else { json!(null) };
let hb = json!({"op": 1, "d": d});
if write.send(Message::Text(hb.to_string())).await.is_err() {
if write
.send(Message::Text(hb.to_string().into()))
.await
.is_err()
{
break;
}
continue;