1use std::{str::FromStr, sync::Arc};23use iroh::endpoint::{self, presets, presets::Preset as _};4use napi::{bindgen_prelude::*, threadsafe_function::ThreadsafeFunction};5use napi_derive::napi;6use tokio::sync::Mutex;78use crate::{EndpointAddr, RelayMode, SecretKey, WatchHandle, path, watch};910/// A mutable handle to an endpoint builder.11///12/// Mirrors the uniffi `EndpointBuilder`. A "preset" in JS is simply any13/// function `(builder: EndpointBuilder) => void` — call one of the built-in14/// [`presetN0`] / [`presetMinimal`] / [`presetN0DisableRelay`] helpers (which15/// install the crypto provider) and then layer on your own configuration.16///17/// ```js18/// const b = Endpoint.builder()19/// presetMinimal(b)20/// b.alpns([alpn])21/// const ep = await b.bind()22/// ```2324pub struct EndpointBuilder {25 inner: std::sync::Mutex<Option<iroh::endpoint::Builder>>,26}2728impl EndpointBuilder {29 fn new(builder: iroh::endpoint::Builder) -> Self {30 Self {31 inner: std::sync::Mutex::new(Some(builder)),32 }33 }3435 fn map<F>(&self, f: F)36 where37 F: FnOnce(iroh::endpoint::Builder) -> iroh::endpoint::Builder,38 {39 let mut guard = self.inner.lock().unwrap();40 let b = guard.take().expect("EndpointBuilder consumed");41 *guard = Some(f(b));42 }43}444546impl EndpointBuilder {47 /// Replay the n0 production preset (relays + discovery + crypto provider).48 49 pub fn apply_n0(&self) {50 self.map(|b| presets::N0.apply(b));51 }5253 /// Replay the minimal preset (crypto provider only, no external deps).54 55 pub fn apply_minimal(&self) {56 self.map(|b| presets::Minimal.apply(b));57 }5859 /// Replay the n0 preset with relays disabled.60 61 pub fn apply_n0_disable_relay(&self) {62 self.map(|b| presets::N0DisableRelay.apply(b));63 }6465 /// Set the endpoint secret key (32 bytes).66 67 pub fn secret_key(&self, bytes: Vec<u8>) -> Result<()> {68 let key: [u8; 32] = bytes69 .try_into()70 .map_err(|_| anyhow::anyhow!("secret_key must be 32 bytes"))?;71 self.map(|b| b.secret_key(iroh::SecretKey::from_bytes(&key)));72 Ok(())73 }7475 /// Set the advertised ALPNs.76 77 pub fn alpns(&self, alpns: Vec<Vec<u8>>) {78 self.map(|b| b.alpns(alpns));79 }8081 /// Set the relay mode.82 83 pub fn relay_mode(&self, mode: &RelayMode) {84 let mode = mode.0.clone();85 self.map(|b| b.relay_mode(mode));86 }8788 /// Set the address the endpoint binds to (`host:port`).89 90 pub fn bind_addr(&self, addr: String) -> Result<()> {91 let socket = std::net::SocketAddr::from_str(&addr).map_err(anyhow::Error::from)?;92 let mut guard = self.inner.lock().unwrap();93 let b = guard.take().expect("EndpointBuilder consumed");94 *guard = Some(b.bind_addr(socket).map_err(anyhow::Error::from)?);95 Ok(())96 }9798 /// Bind the endpoint.99 100 pub async fn bind(&self) -> Result<Endpoint> {101 let builder = self102 .inner103 .lock()104 .unwrap()105 .take()106 .ok_or_else(|| anyhow::anyhow!("EndpointBuilder already consumed"))?;107 let endpoint = builder.bind().await.map_err(|e| anyhow::anyhow!("{e:?}"))?;108 Ok(Endpoint { inner: endpoint })109 }110}111112/// The n0 production preset (relays + discovery).113114pub fn preset_n0(builder: &EndpointBuilder) {115 builder.apply_n0();116}117118/// The minimal preset (no external dependencies; good for tests / offline).119120pub fn preset_minimal(builder: &EndpointBuilder) {121 builder.apply_minimal();122}123124/// The n0 preset with relays disabled.125126pub fn preset_n0_disable_relay(builder: &EndpointBuilder) {127 builder.apply_n0_disable_relay();128}129130/// Which side of a connection we are.131132133pub enum Side {134 Client,135 Server,136}137138impl From<iroh::endpoint::Side> for Side {139 fn from(s: iroh::endpoint::Side) -> Self {140 match s {141 iroh::endpoint::Side::Client => Side::Client,142 iroh::endpoint::Side::Server => Side::Server,143 }144 }145}146147/// Where an incoming connection came from.148149150pub struct IncomingAddr {151 /// One of "ip" | "relay" | "custom".152 pub kind: String,153 /// `ip:port` for ip, relay URL for relay.154 pub addr: Option<String>,155 /// Remote endpoint id (relay only).156 pub endpoint_id: Option<String>,157 /// Debug description (custom only).158 pub description: Option<String>,159}160161impl From<iroh::endpoint::IncomingAddr> for IncomingAddr {162 fn from(addr: iroh::endpoint::IncomingAddr) -> Self {163 match addr {164 iroh::endpoint::IncomingAddr::Ip(socket) => IncomingAddr {165 kind: "ip".into(),166 addr: Some(socket.to_string()),167 endpoint_id: None,168 description: None,169 },170 iroh::endpoint::IncomingAddr::Relay { url, endpoint_id } => IncomingAddr {171 kind: "relay".into(),172 addr: Some(url.to_string()),173 endpoint_id: Some(endpoint_id.to_string()),174 description: None,175 },176 iroh::endpoint::IncomingAddr::Custom(c) => IncomingAddr {177 kind: "custom".into(),178 addr: None,179 endpoint_id: None,180 description: Some(format!("{c:?}")),181 },182 _ => IncomingAddr {183 kind: "custom".into(),184 addr: None,185 endpoint_id: None,186 description: Some("unknown".into()),187 },188 }189 }190}191192/// The local address that received an incoming connection.193194195pub struct IncomingLocalAddr {196 /// One of "ip" | "relay" | "custom".197 pub kind: String,198 pub addr: Option<String>,199 pub description: Option<String>,200}201202impl From<iroh::endpoint::LocalTransportAddr> for IncomingLocalAddr {203 fn from(value: iroh::endpoint::LocalTransportAddr) -> Self {204 match value {205 iroh::endpoint::LocalTransportAddr::Ip(ip) => IncomingLocalAddr {206 kind: "ip".into(),207 addr: ip.map(|i| i.to_string()),208 description: None,209 },210 iroh::endpoint::LocalTransportAddr::Relay(url) => IncomingLocalAddr {211 kind: "relay".into(),212 addr: Some(url.to_string()),213 description: None,214 },215 iroh::endpoint::LocalTransportAddr::Custom(c) => IncomingLocalAddr {216 kind: "custom".into(),217 addr: None,218 description: c.map(|c| format!("{c:?}")),219 },220 _ => IncomingLocalAddr {221 kind: "custom".into(),222 addr: None,223 description: None,224 },225 }226 }227}228229/// A snapshot value for a single endpoint metric.230231232pub struct CounterStats {233 pub value: u32,234 pub description: String,235}236237/// Flat snapshot of headline connection statistics.238239240pub struct ConnectionStats {241 pub udp_tx_datagrams: i64,242 pub udp_tx_bytes: i64,243 pub udp_rx_datagrams: i64,244 pub udp_rx_bytes: i64,245 pub lost_packets: i64,246 pub lost_bytes: i64,247}248249/// Options passed to [`Endpoint::bind`].250///251/// `bind` applies the n0 preset by default. For a custom preset use252/// [`Endpoint::builder`] + the `EndpointBuilder` surface.253254255pub struct EndpointOptions {256 pub bind_addr: Option<String>,257 pub secret_key: Option<Vec<u8>>,258 pub alpns: Option<Vec<Vec<u8>>>,259}260261/// An iroh endpoint.262263pub struct Endpoint {264 inner: endpoint::Endpoint,265}266267impl Endpoint {268 pub(crate) fn raw(&self) -> &endpoint::Endpoint {269 &self.inner270 }271}272273274impl Endpoint {275 /// Create an endpoint builder (starts empty — apply a preset).276 277 pub fn builder() -> EndpointBuilder {278 EndpointBuilder::new(iroh::endpoint::Builder::empty())279 }280281 /// Bind a new endpoint. Applies the n0 preset, then the given options.282 /// For a custom preset use [`Endpoint::builder`].283 284 pub async fn bind(285 options: Option<EndpointOptions>,286 relay_mode: Option<&RelayMode>,287 ) -> Result<Self> {288 let options = options.unwrap_or_default();289 let wrapper = EndpointBuilder::new(iroh::endpoint::Builder::empty());290 wrapper.apply_n0();291292 if let Some(secret_key) = options.secret_key {293 wrapper.secret_key(secret_key)?;294 }295 if let Some(alpns) = options.alpns {296 wrapper.alpns(alpns);297 }298 if let Some(relay_mode) = relay_mode {299 wrapper.relay_mode(relay_mode);300 }301 if let Some(addr) = options.bind_addr {302 wrapper.bind_addr(addr)?;303 }304305 wrapper.bind().await306 }307308 /// This endpoint's id.309 310 pub fn id(&self) -> crate::EndpointId {311 self.inner.id().into()312 }313314 /// The [`EndpointAddr`] for this endpoint.315 316 pub fn addr(&self) -> EndpointAddr {317 self.inner.addr().into()318 }319320 /// Look up cached information about a remote endpoint, if any.321 322 pub async fn remote_addr(&self, id: &crate::EndpointId) -> Result<Option<EndpointAddr>> {323 let id: iroh::EndpointId = id.into();324 let info = match self.inner.remote_info(id).await {325 Some(i) => i,326 None => return Ok(None),327 };328 let id = info.id();329 let addrs = info.into_addrs().map(|a| a.into_addr());330 Ok(Some(iroh::EndpointAddr::from_parts(id, addrs).into()))331 }332333 /// Current statistics for this endpoint.334 335 pub fn stats(&self) -> std::collections::HashMap<String, CounterStats> {336 use iroh_metrics::{MetricValue, MetricsGroupSet};337 self.inner338 .metrics()339 .iter()340 .map(|(group, item)| {341 let name = format!("{}:{}", group, item.name());342 let value = match item.value() {343 MetricValue::Counter(v) => u32::try_from(v).unwrap_or(u32::MAX),344 MetricValue::Gauge(v) => u32::try_from(v.max(0)).unwrap_or(u32::MAX),345 _ => 0,346 };347 (348 name,349 CounterStats {350 value,351 description: item.help().to_string(),352 },353 )354 })355 .collect()356 }357358 /// The secret key backing this endpoint's identity.359 360 pub fn secret_key(&self) -> SecretKey {361 self.inner.secret_key().clone().into()362 }363364 /// Replace the set of advertised ALPNs.365 366 pub fn set_alpns(&self, alpns: Vec<Vec<u8>>) {367 self.inner.set_alpns(alpns);368 }369370 /// Add an external (manually-known) socket address.371 372 pub async fn add_external_addr(&self, addr: String) -> Result<()> {373 let socket = std::net::SocketAddr::from_str(&addr).map_err(anyhow::Error::from)?;374 self.inner.add_external_addr(socket).await;375 Ok(())376 }377378 /// Remove a previously-added external address.379 380 pub async fn remove_external_addr(&self, addr: String) -> Result<bool> {381 let socket = std::net::SocketAddr::from_str(&addr).map_err(anyhow::Error::from)?;382 Ok(self.inner.remove_external_addr(&socket).await)383 }384385 /// The local socket addresses this endpoint is bound to.386 387 pub fn bound_sockets(&self) -> Vec<String> {388 self.inner389 .bound_sockets()390 .into_iter()391 .map(|a| a.to_string())392 .collect()393 }394395 /// Resolves once the endpoint has a usable home relay.396 397 pub async fn online(&self) {398 self.inner.online().await;399 }400401 /// Insert (or replace) a relay configuration at runtime.402 403 pub async fn insert_relay(&self, config: crate::RelayConfig) -> Result<()> {404 let config: iroh::RelayConfig = config.try_into()?;405 let url = config.url.clone();406 self.inner.insert_relay(url, Arc::new(config)).await;407 Ok(())408 }409410 /// Remove a relay configuration at runtime.411 412 pub async fn remove_relay(&self, url: String) -> Result<bool> {413 let url = iroh::RelayUrl::from_str(&url).map_err(anyhow::Error::from)?;414 Ok(self.inner.remove_relay(&url).await.is_some())415 }416417 /// Connect to a remote endpoint via the given ALPN.418 419 pub async fn connect(&self, addr: &EndpointAddr, alpn: Vec<u8>) -> Result<Connection> {420 let addr: iroh::EndpointAddr = addr.try_into()?;421 let conn = self422 .inner423 .connect(addr, &alpn)424 .await425 .map_err(|e| anyhow::anyhow!("{e:?}"))?;426 Ok(Connection(Arc::new(conn)))427 }428429 /// Begin a connection attempt, returning the in-progress handle.430 431 pub async fn connect_pending(&self, addr: &EndpointAddr, alpn: Vec<u8>) -> Result<Connecting> {432 let addr: iroh::EndpointAddr = addr.try_into()?;433 let connecting = self434 .inner435 .connect_with_opts(addr, &alpn, iroh::endpoint::ConnectOptions::default())436 .await437 .map_err(|e| anyhow::anyhow!("{e:?}"))?;438 Ok(Connecting(Mutex::new(Some(connecting))))439 }440441 /// Pull the next incoming connection attempt.442 443 pub async fn accept_next(&self) -> Option<Incoming> {444 let inner = self.inner.accept().await?;445 Some(Incoming(Mutex::new(Some(inner))))446 }447448 /// Watch for changes to this endpoint's address.449 450 pub fn watch_addr(&self, callback: ThreadsafeFunction<EndpointAddr>) -> WatchHandle {451 watch::spawn_watch_addr(self.inner.clone(), callback)452 }453454 /// Watch for changes to the connected relays.455 456 pub fn watch_home_relay(&self, callback: ThreadsafeFunction<Vec<String>>) -> WatchHandle {457 watch::spawn_home_relay_watch(self.inner.clone(), callback)458 }459460 /// Watch for network-stack changes.461 462 pub fn watch_network_change(&self, callback: ThreadsafeFunction<()>) -> WatchHandle {463 watch::spawn_network_change_watch(self.inner.clone(), callback)464 }465466 /// Close the endpoint.467 468 pub async fn close(&self) {469 self.inner.close().await;470 }471472 /// True if the endpoint is closed.473 474 pub fn is_closed(&self) -> bool {475 self.inner.is_closed()476 }477}478479/// An incoming connection that has not yet begun its server-side handshake.480481pub struct Incoming(Mutex<Option<iroh::endpoint::Incoming>>);482483484impl Incoming {485 486 pub async fn accept(&self) -> Result<Accepting> {487 let inner = self488 .0489 .lock()490 .await491 .take()492 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;493 let accepting = inner.accept().map_err(|e| anyhow::anyhow!("{e:?}"))?;494 Ok(Accepting(Mutex::new(Some(accepting))))495 }496497 498 pub async fn refuse(&self) -> Result<()> {499 let inner = self500 .0501 .lock()502 .await503 .take()504 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;505 inner.refuse();506 Ok(())507 }508509 510 pub async fn retry(&self) -> Result<()> {511 let inner = self512 .0513 .lock()514 .await515 .take()516 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;517 inner518 .retry()519 .map_err(|e| anyhow::anyhow!("retry failed: {e:?}").into())520 }521522 523 pub async fn ignore(&self) -> Result<()> {524 let inner = self525 .0526 .lock()527 .await528 .take()529 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;530 inner.ignore();531 Ok(())532 }533534 535 pub async fn local_addr(&self) -> Result<IncomingLocalAddr> {536 let guard = self.0.lock().await;537 let inner = guard538 .as_ref()539 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;540 Ok(inner.local_addr().into())541 }542543 544 pub async fn remote_addr(&self) -> Result<IncomingAddr> {545 let guard = self.0.lock().await;546 let inner = guard547 .as_ref()548 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;549 Ok(inner.remote_addr().into())550 }551552 553 pub async fn remote_addr_validated(&self) -> Result<bool> {554 let guard = self.0.lock().await;555 let inner = guard556 .as_ref()557 .ok_or_else(|| anyhow::anyhow!("Incoming already consumed"))?;558 Ok(inner.remote_addr_validated())559 }560}561562/// A server-side handshake in progress.563564pub struct Accepting(Mutex<Option<iroh::endpoint::Accepting>>);565566567impl Accepting {568 569 pub async fn connect(&self) -> Result<Connection> {570 let inner = self571 .0572 .lock()573 .await574 .take()575 .ok_or_else(|| anyhow::anyhow!("Accepting already consumed"))?;576 let conn = inner.await.map_err(|e| anyhow::anyhow!("{e:?}"))?;577 Ok(Connection(Arc::new(conn)))578 }579580 581 pub async fn alpn(&self) -> Result<Vec<u8>> {582 let mut guard = self.0.lock().await;583 let inner = guard584 .as_mut()585 .ok_or_else(|| anyhow::anyhow!("Accepting already consumed"))?;586 inner587 .alpn()588 .await589 .map_err(|e| anyhow::anyhow!("{e:?}").into())590 }591}592593/// A client-side handshake in progress.594595pub struct Connecting(Mutex<Option<iroh::endpoint::Connecting>>);596597598impl Connecting {599 600 pub async fn connect(&self) -> Result<Connection> {601 let inner = self602 .0603 .lock()604 .await605 .take()606 .ok_or_else(|| anyhow::anyhow!("Connecting already consumed"))?;607 let conn = inner.await.map_err(|e| anyhow::anyhow!("{e:?}"))?;608 Ok(Connection(Arc::new(conn)))609 }610611 612 pub async fn alpn(&self) -> Result<Vec<u8>> {613 let mut guard = self.0.lock().await;614 let inner = guard615 .as_mut()616 .ok_or_else(|| anyhow::anyhow!("Connecting already consumed"))?;617 inner618 .alpn()619 .await620 .map_err(|e| anyhow::anyhow!("{e:?}").into())621 }622623 624 pub async fn remote_id(&self) -> Result<crate::EndpointId> {625 let guard = self.0.lock().await;626 let inner = guard627 .as_ref()628 .ok_or_else(|| anyhow::anyhow!("Connecting already consumed"))?;629 Ok(inner.remote_id().into())630 }631}632633/// An active QUIC connection.634635pub struct Connection(Arc<endpoint::Connection>);636637638impl Connection {639 640 pub fn alpn(&self) -> Vec<u8> {641 self.0.alpn().to_vec()642 }643644 645 pub fn remote_id(&self) -> crate::EndpointId {646 self.0.remote_id().into()647 }648649 650 pub fn side(&self) -> Side {651 self.0.side().into()652 }653654 655 pub async fn open_uni(&self) -> Result<SendStream> {656 let s = self657 .0658 .open_uni()659 .await660 .map_err(|e| anyhow::anyhow!("{e:?}"))?;661 Ok(SendStream::new(s))662 }663664 665 pub async fn accept_uni(&self) -> Result<RecvStream> {666 let r = self667 .0668 .accept_uni()669 .await670 .map_err(|e| anyhow::anyhow!("{e:?}"))?;671 Ok(RecvStream::new(r))672 }673674 675 pub async fn open_bi(&self) -> Result<BiStream> {676 let (s, r) = self677 .0678 .open_bi()679 .await680 .map_err(|e| anyhow::anyhow!("{e:?}"))?;681 Ok(BiStream {682 send: SendStream::new(s),683 recv: RecvStream::new(r),684 })685 }686687 688 pub async fn accept_bi(&self) -> Result<BiStream> {689 let (s, r) = self690 .0691 .accept_bi()692 .await693 .map_err(|e| anyhow::anyhow!("{e:?}"))?;694 Ok(BiStream {695 send: SendStream::new(s),696 recv: RecvStream::new(r),697 })698 }699700 701 pub async fn read_datagram(&self) -> Result<Vec<u8>> {702 let res = self703 .0704 .read_datagram()705 .await706 .map_err(|e| anyhow::anyhow!("{e:?}"))?;707 Ok(res.to_vec())708 }709710 711 pub async fn closed(&self) -> String {712 self.0.closed().await.to_string()713 }714715 716 pub fn close_reason(&self) -> Option<String> {717 self.0.close_reason().map(|s| s.to_string())718 }719720 721 pub fn close(&self, error_code: BigInt, reason: Vec<u8>) -> Result<()> {722 let code =723 endpoint::VarInt::from_u64(error_code.get_u64().1).map_err(anyhow::Error::from)?;724 self.0.close(code, &reason);725 Ok(())726 }727728 729 pub fn send_datagram(&self, data: Vec<u8>) -> Result<()> {730 self.0731 .send_datagram(data.into())732 .map_err(|e| anyhow::anyhow!("{e:?}"))?;733 Ok(())734 }735736 737 pub async fn send_datagram_wait(&self, data: Vec<u8>) -> Result<()> {738 self.0739 .send_datagram_wait(data.into())740 .await741 .map_err(|e| anyhow::anyhow!("{e:?}"))?;742 Ok(())743 }744745 746 pub fn max_datagram_size(&self) -> Option<i64> {747 self.0.max_datagram_size().map(|s| s as i64)748 }749750 751 pub fn datagram_send_buffer_space(&self) -> i64 {752 self.0.datagram_send_buffer_space() as i64753 }754755 756 pub fn stable_id(&self) -> u32 {757 self.0.stable_id() as _758 }759760 761 pub fn rtt(&self) -> Option<i64> {762 self.0763 .paths()764 .iter()765 .find(|p| p.is_selected())766 .map(|p| p.rtt().as_millis() as i64)767 }768769 770 pub fn stats(&self) -> ConnectionStats {771 let s = self.0.stats();772 ConnectionStats {773 udp_tx_datagrams: s.udp_tx.datagrams as i64,774 udp_tx_bytes: s.udp_tx.bytes as i64,775 udp_rx_datagrams: s.udp_rx.datagrams as i64,776 udp_rx_bytes: s.udp_rx.bytes as i64,777 lost_packets: s.lost_packets as i64,778 lost_bytes: s.lost_bytes as i64,779 }780 }781782 783 pub fn paths(&self) -> Vec<path::PathSnapshot> {784 path::snapshot_paths(&self.0)785 }786787 788 pub fn watch_paths(789 &self,790 callback: ThreadsafeFunction<Vec<path::PathSnapshot>>,791 ) -> WatchHandle {792 watch::spawn_paths_watch((*self.0).clone(), callback)793 }794795 796 pub fn watch_path_events(&self, callback: ThreadsafeFunction<path::PathEvent>) -> WatchHandle {797 watch::spawn_path_events_watch((*self.0).clone(), callback)798 }799800 801 pub fn set_max_concurrent_uni_streams(&self, count: BigInt) -> Result<()> {802 let n = endpoint::VarInt::from_u64(count.get_u64().1).map_err(anyhow::Error::from)?;803 self.0.set_max_concurrent_uni_streams(n);804 Ok(())805 }806807 808 pub fn set_receive_window(&self, count: BigInt) -> Result<()> {809 let n = endpoint::VarInt::from_u64(count.get_u64().1).map_err(anyhow::Error::from)?;810 self.0.set_receive_window(n);811 Ok(())812 }813814 815 pub fn set_max_concurrent_bi_streams(&self, count: BigInt) -> Result<()> {816 let n = endpoint::VarInt::from_u64(count.get_u64().1).map_err(anyhow::Error::from)?;817 self.0.set_max_concurrent_bi_streams(n);818 Ok(())819 }820}821822/// A bidirectional QUIC stream pair.823824pub struct BiStream {825 send: SendStream,826 recv: RecvStream,827}828829830impl BiStream {831 832 pub fn send(&self) -> SendStream {833 self.send.clone()834 }835836 837 pub fn recv(&self) -> RecvStream {838 self.recv.clone()839 }840}841842/// The outgoing half of a QUIC stream.843844845pub struct SendStream(pub(crate) Arc<Mutex<endpoint::SendStream>>); // lofi-node: pub(crate) for lofi_ext framing846847impl SendStream {848 fn new(s: endpoint::SendStream) -> Self {849 SendStream(Arc::new(Mutex::new(s)))850 }851}852853854impl SendStream {855 856 pub async fn write(&self, buf: Vec<u8>) -> Result<i64> {857 let mut s = self.0.lock().await;858 let n = s.write(&buf).await.map_err(|e| anyhow::anyhow!("{e:?}"))?;859 Ok(n as i64)860 }861862 863 pub async fn write_all(&self, buf: Vec<u8>) -> Result<()> {864 let mut s = self.0.lock().await;865 s.write_all(&buf)866 .await867 .map_err(|e| anyhow::anyhow!("{e:?}"))?;868 Ok(())869 }870871 872 pub async fn finish(&self) -> Result<()> {873 let mut s = self.0.lock().await;874 s.finish().map_err(|e| anyhow::anyhow!("{e:?}"))?;875 Ok(())876 }877878 879 pub async fn reset(&self, error_code: BigInt) -> Result<()> {880 let code =881 endpoint::VarInt::from_u64(error_code.get_u64().1).map_err(anyhow::Error::from)?;882 let mut s = self.0.lock().await;883 s.reset(code).map_err(|e| anyhow::anyhow!("{e:?}"))?;884 Ok(())885 }886887 888 pub async fn set_priority(&self, p: i32) -> Result<()> {889 let s = self.0.lock().await;890 s.set_priority(p).map_err(|e| anyhow::anyhow!("{e:?}"))?;891 Ok(())892 }893894 895 pub async fn priority(&self) -> Result<i32> {896 let s = self.0.lock().await;897 s.priority().map_err(|e| anyhow::anyhow!("{e:?}").into())898 }899900 901 pub async fn stopped(&self) -> Result<Option<i64>> {902 let s = self.0.lock().await;903 let res = s.stopped().await.map_err(|e| anyhow::anyhow!("{e:?}"))?;904 Ok(res.map(|r| r.into_inner() as i64))905 }906907 908 pub async fn id(&self) -> String {909 self.0.lock().await.id().to_string()910 }911}912913/// The incoming half of a QUIC stream.914915916pub struct RecvStream(pub(crate) Arc<Mutex<endpoint::RecvStream>>); // lofi-node: pub(crate) for lofi_ext framing917918impl RecvStream {919 fn new(s: endpoint::RecvStream) -> Self {920 RecvStream(Arc::new(Mutex::new(s)))921 }922}923924925impl RecvStream {926 927 pub async fn read(&self, size_limit: u32) -> Result<Vec<u8>> {928 let mut buf = vec![0u8; size_limit as usize];929 let mut r = self.0.lock().await;930 let res = r931 .read(&mut buf)932 .await933 .map_err(|e| anyhow::anyhow!("{e:?}"))?;934 let len = res.unwrap_or(0);935 buf.truncate(len);936 Ok(buf)937 }938939 940 pub async fn read_exact(&self, size: u32) -> Result<Vec<u8>> {941 let mut buf = vec![0u8; size as usize];942 let mut r = self.0.lock().await;943 r.read_exact(&mut buf)944 .await945 .map_err(|e| anyhow::anyhow!("{e:?}"))?;946 Ok(buf)947 }948949 950 pub async fn read_to_end(&self, size_limit: u32) -> Result<Vec<u8>> {951 let mut r = self.0.lock().await;952 let res = r953 .read_to_end(size_limit as usize)954 .await955 .map_err(|e| anyhow::anyhow!("{e:?}"))?;956 Ok(res)957 }958959 960 pub async fn id(&self) -> String {961 self.0.lock().await.id().to_string()962 }963964 965 pub async fn bytes_read(&self) -> Result<i64> {966 let r = self.0.lock().await;967 Ok(r.bytes_read().map_err(|e| anyhow::anyhow!("{e:?}"))? as i64)968 }969970 971 pub async fn stop(&self, error_code: BigInt) -> Result<()> {972 let code =973 endpoint::VarInt::from_u64(error_code.get_u64().1).map_err(anyhow::Error::from)?;974 let mut r = self.0.lock().await;975 r.stop(code).map_err(|e| anyhow::anyhow!("{e:?}"))?;976 Ok(())977 }978979 980 pub async fn received_reset(&self) -> Result<Option<i64>> {981 let mut r = self.0.lock().await;982 let code = r983 .received_reset()984 .await985 .map_err(|e| anyhow::anyhow!("{e:?}"))?;986 Ok(code.map(|c| c.into_inner() as i64))987 }988}