1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
use crate::storage::collection::ChangeEvent;
use futures::{SinkExt, StreamExt};
use tokio_tungstenite::{
connect_async,
tungstenite::{client::IntoClientRequest, protocol::Message},
};
use url::Url;
/// Client for connecting to other nodes' WebSocket changefeeds
pub struct ClusterWebsocketClient;
impl ClusterWebsocketClient {
/// Connect to a remote node's changefeed and return a stream of ChangeEvents
///
/// # Arguments
/// * `node_addr` - Address of the remote node
/// * `database` - Database name
/// * `collection` - Collection name
/// * `local_only` - Whether to get only local changes
/// * `cluster_secret` - Cluster secret from keyfile for authentication
pub async fn connect(
node_addr: &str,
database: &str,
collection: &str,
local_only: bool,
cluster_secret: &str,
) -> anyhow::Result<impl futures::Stream<Item = anyhow::Result<ChangeEvent>>> {
// Construct WebSocket URL with cluster-internal authentication
if cluster_secret.is_empty() {
return Err(anyhow::anyhow!(
"Cluster secret not configured - cannot connect to cluster WebSocket"
));
}
let url_str = format!(
"ws://{}/_api/ws/changefeed?token=cluster-internal",
node_addr
);
let url = Url::parse(&url_str)?;
tracing::debug!(
"[CLUSTER-WS] Connecting to {} (local_only={})",
url,
local_only
);
// Connect with cluster secret header for authentication
let mut request = IntoClientRequest::into_client_request(url.as_str())?;
request
.headers_mut()
.insert("X-Cluster-Secret", cluster_secret.parse().unwrap());
let (ws_stream, _) = connect_async(request).await?;
let (mut write, mut read) = ws_stream.split();
// Send subscription message
let subscribe_msg = serde_json::json!({
"type": "subscribe",
"database": database,
"collection": collection,
"local": local_only
});
write
.send(Message::Text(subscribe_msg.to_string().into()))
.await?;
// Return a stream that parses messages
let stream = async_stream::try_stream! {
while let Some(msg) = read.next().await {
match msg {
Ok(Message::Text(text)) => {
// Skip "subscribed" confirmation or errors for now, just try to parse event
if let Ok(event) = serde_json::from_str::<ChangeEvent>(&text) {
yield event;
} else {
// Might be a control message like {"type": "subscribed"}
tracing::trace!("[CLUSTER-WS] Received non-event message: {}", text);
}
}
Ok(Message::Close(_)) => break,
Err(e) => Err(anyhow::anyhow!("WebSocket error: {}", e))?,
_ => {}
}
}
};
Ok(stream)
}
}