#[error("tcp layer dropped")]
TcpDropped,
#[error("future cursor: {message:?}")]
FutureCursor { message: Option<String> },
#[error("connect timed out")]
ConnectTimeout,
}
impl From<serde_ipld_dagcbor::DecodeError<Infallible>> for FirehoseError {
fn from(e: serde_ipld_dagcbor::DecodeError<Infallible>) -> Self {
/// time and starves the runtime when thousands of PDS streams start at once.
/// if it can't be built, connects fall back to that path and report its error.
static TLS_CONNECTOR: LazyLock<Option<Connector>> = LazyLock::new(|| Connector::new().ok());
/// a host that takes the tcp connection and never answers the upgrade would park the task forever
const CONNECT_TIMEOUT: Duration = if cfg!(test) {
Duration::from_millis(200)
} else {
Duration::from_secs(30)
};
pub struct FirehoseStream {
ws: WebSocketStream<tokio_websockets::MaybeTlsStream<tokio::net::TcpStream>>,
ping_timer: tokio::time::Interval,
}
&& let Some(connector) = TLS_CONNECTOR.as_ref()
{
builder = builder.connector(connector);
}
let connect = async {
let (ws, _) = if public {
builder.resolver(PublicResolver::system()).connect().await?
} else {
builder.connect().await?
if public {
builder.resolver(PublicResolver::system()).connect().await
} else {
builder.connect().await
}
};
let (ws, _) = tokio::time::timeout(CONNECT_TIMEOUT, connect)
.await
.map_err(|_| FirehoseError::ConnectTimeout)??;
let mut ping_timer = tokio::time::interval(Duration::from_secs(45));
ping_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
ping_timer.tick().await; // consume the first tick
199 unmodified lines
Expand all
assert!(
matches!(result, Err(FirehoseError::InvalidUri(message)) if message.contains("non-public IP"))
);
}
#[tokio::test]
async fn connect_gives_up_on_a_host_that_never_answers_the_upgrade() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = Url::parse(&format!("ws://{}/", listener.local_addr().unwrap())).unwrap();
let held = tokio::spawn(async move {
let _conn = listener.accept().await;
std::future::pending::<()>().await
});
let res = FirehoseStream::connect(url, None, false).await;
assert!(matches!(res, Err(FirehoseError::ConnectTimeout)));
held.abort();
}
}