110 lines
3.7 KiB
Rust
110 lines
3.7 KiB
Rust
mod common;
|
|
|
|
use std::time::Duration;
|
|
|
|
use common::{
|
|
challenge, client, publish, query_envelope, register, spawn_relay, spawn_relay_with_capacity,
|
|
};
|
|
use frxd::crypto::Keypair;
|
|
use futures_util::StreamExt;
|
|
|
|
async fn open_stream(http: &reqwest::Client, relay: &str, key: &Keypair) -> reqwest::Response {
|
|
let member = key.public_hex();
|
|
let nonce = challenge(http, relay, &member).await.expect("challenge");
|
|
let sig = key.sign(&frxd::crypto::poll_signing_bytes(&member, &nonce));
|
|
http.get(format!(
|
|
"{relay}/v1/stream?member={member}&nonce={nonce}&sig={sig}"
|
|
))
|
|
.send()
|
|
.await
|
|
.unwrap()
|
|
}
|
|
|
|
async fn read_until<S>(mut stream: S, needle: &str, timeout: Duration) -> (String, S)
|
|
where
|
|
S: futures_util::Stream<Item = Result<bytes::Bytes, reqwest::Error>> + Unpin,
|
|
{
|
|
let mut buffer = String::new();
|
|
let deadline = tokio::time::Instant::now() + timeout;
|
|
loop {
|
|
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
|
|
let chunk = tokio::time::timeout(remaining, stream.next())
|
|
.await
|
|
.expect("timed out waiting for stream data");
|
|
match chunk {
|
|
Some(Ok(bytes)) => {
|
|
buffer.push_str(&String::from_utf8_lossy(&bytes));
|
|
if buffer.contains(needle) {
|
|
return (buffer, stream);
|
|
}
|
|
}
|
|
Some(Err(error)) => panic!("stream error: {error}"),
|
|
None => panic!("stream closed before '{needle}'"),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn stream_delivers_envelopes_live() {
|
|
let relay = spawn_relay().await;
|
|
let http = client();
|
|
let member = Keypair::generate();
|
|
let publisher = Keypair::generate();
|
|
|
|
let response = open_stream(&http, &relay, &member).await;
|
|
assert!(response.status().is_success());
|
|
assert!(
|
|
response
|
|
.headers()
|
|
.get("content-type")
|
|
.and_then(|value| value.to_str().ok())
|
|
.unwrap_or("")
|
|
.contains("text/event-stream")
|
|
);
|
|
|
|
let envelope = query_envelope(&publisher, &publisher.public_hex(), "live stream", 5);
|
|
assert!(
|
|
publish(&http, &relay, &envelope)
|
|
.await
|
|
.status()
|
|
.is_success()
|
|
);
|
|
|
|
let (buffer, _) = read_until(
|
|
response.bytes_stream(),
|
|
"live stream",
|
|
Duration::from_secs(3),
|
|
)
|
|
.await;
|
|
assert!(buffer.contains("event: envelope"), "{buffer}");
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
|
|
async fn stream_reports_lag_then_recovers() {
|
|
let relay = spawn_relay_with_capacity(1).await;
|
|
let http = client();
|
|
let member = Keypair::generate();
|
|
let publisher = Keypair::generate();
|
|
register(&http, &relay, &member).await;
|
|
|
|
let first = query_envelope(&publisher, &publisher.public_hex(), "first", 5);
|
|
assert!(publish(&http, &relay, &first).await.status().is_success());
|
|
let second = query_envelope(&publisher, &publisher.public_hex(), "second", 5);
|
|
assert!(publish(&http, &relay, &second).await.status().is_success());
|
|
|
|
let response = open_stream(&http, &relay, &member).await;
|
|
assert!(response.status().is_success());
|
|
let (buffer, stream) = read_until(
|
|
response.bytes_stream(),
|
|
"\"missed\"",
|
|
Duration::from_secs(3),
|
|
)
|
|
.await;
|
|
assert!(buffer.contains("event: lag"), "{buffer}");
|
|
|
|
let third = query_envelope(&publisher, &publisher.public_hex(), "third", 5);
|
|
assert!(publish(&http, &relay, &third).await.status().is_success());
|
|
let (buffer, _) = read_until(stream, "third", Duration::from_secs(3)).await;
|
|
assert!(buffer.contains("event: envelope"), "{buffer}");
|
|
}
|