summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/transport/identity.rs313
-rw-r--r--src/transport/mod.rs1
-rw-r--r--src/transport/paths.rs364
3 files changed, 428 insertions, 250 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs
index 17c7795..7633ac1 100644
--- a/src/transport/identity.rs
+++ b/src/transport/identity.rs
@@ -22,22 +22,22 @@
//! with only hints, those are asked WHO, and the send is refused until one answers
//! (the node retries refused sends).
//!
-//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`, or
-//! `<scheme>:<location>` where a transport needs more than an identity to reach a node
-//! (`Transport::locate`, `Transport::locations`); the books keep the identities.
+//! Addresses and the transports behind them are the paths layer's
+//! (`transport::paths`): the books keep addresses as paths keep them.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, MutexGuard};
use std::time::Duration;
-use anyhow::{Context, Result, anyhow, bail};
+use anyhow::{Result, anyhow, bail};
use async_trait::async_trait;
-use tokio::sync::{mpsc, oneshot};
+use tokio::sync::oneshot;
use tokio::time::Instant;
use tracing::{debug, info};
use crate::identity::{self, NodeKey};
use crate::locks::MutexExt;
+use crate::transport::paths::{PathHandler, Paths, scheme_of};
use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats};
use crate::transport::traffic::Traffic;
use crate::transport::{
@@ -94,25 +94,8 @@ pub(crate) struct IdentityTransport {
shared: Arc<Shared>,
}
-/// One of the node's transports.
-struct Link {
- scheme: &'static str,
- inner: Arc<dyn Transport>,
-}
-
-impl Link {
- fn address(&self, target: &str) -> String {
- format!("{}:{target}", self.scheme)
- }
-
- fn own_address(&self) -> String {
- self.address(self.inner.local_id())
- }
-}
-
struct Shared {
- /// In order of preference.
- links: Vec<Link>,
+ paths: Paths,
key: Arc<NodeKey>,
node_id: String,
/// WHO and ME are counted here, under the address (`traffic`).
@@ -138,11 +121,6 @@ struct Book {
counts: IdentityCounts,
}
-/// The scheme of `address`.
-fn scheme_of(address: &str) -> &str {
- address.split_once(':').map_or("", |(scheme, _)| scheme)
-}
-
impl Book {
/// `node` proved itself at `address`. Returns whether that is news.
fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool {
@@ -236,13 +214,14 @@ impl IdentityTransport {
traffic: Arc<Traffic>,
) -> Self {
let node_id = config.key.id();
- let links = std::iter::once((config.scheme, transport))
- .chain(config.more)
- .map(|(scheme, inner)| Link { scheme, inner })
- .collect();
+ let paths = Paths::new(
+ std::iter::once((config.scheme, transport))
+ .chain(config.more)
+ .collect(),
+ );
Self {
shared: Arc::new(Shared {
- links,
+ paths,
key: config.key,
node_id,
traffic,
@@ -257,66 +236,19 @@ impl Shared {
self.book.lock_or_recover()
}
- /// The transport `address` is on, and the transport identity at it
- /// (`Transport::locate`, which notes where that is if the address says); None if no
- /// transport of ours has its scheme, or it isn't one of that transport's.
- fn split(&self, address: &str) -> Option<(&Link, String)> {
- let (scheme, target) = address.split_once(':')?;
- let link = self.links.iter().find(|link| link.scheme == scheme)?;
- let target = link.inner.locate(target)?;
- Some((link, target))
- }
-
- /// `address` as the books keep it: `<scheme>:<transport identity>`.
- fn canonical(&self, address: &str) -> Option<String> {
- self.split(address)
- .map(|(link, target)| link.address(&target))
- }
-
- /// Where the node at `address` (as the books keep it) is, as others can use it
- /// (`Transport::locations`).
- fn locations(&self, address: &str) -> Vec<String> {
- let Some((link, target)) = self.split(address) else {
- return Vec::new();
- };
- link.inner
- .locations(&target)
- .iter()
- .map(|location| link.address(location))
- .collect()
- }
-
- /// Where `address`'s transport is in the order of preference (none of ours: last).
- fn preference(&self, address: &str) -> usize {
- let scheme = scheme_of(address);
- self.links
- .iter()
- .position(|link| link.scheme == scheme)
- .unwrap_or(usize::MAX)
- }
-
- fn is_own(&self, address: &str) -> bool {
- self.links.iter().any(|link| link.own_address() == address)
- }
-
- fn proof(&self, label: &[u8], link: &Link, target: &str) -> IdentityProof {
- let message = signed(label, &link.own_address(), &link.address(target));
- IdentityProof {
+ /// Our proof for `address` (as paths keep it), from our address on its transport.
+ fn proof(&self, label: &[u8], address: &str) -> Option<IdentityProof> {
+ let message = signed(label, &self.paths.own_address(address)?, address);
+ Some(IdentityProof {
key: self.key.public(),
signature: self.key.sign(&message),
- }
+ })
}
- /// Notes the node `proof` proves to be at `address` (on `link`); None if it proves
- /// nothing.
- fn prove(
- &self,
- link: &Link,
- address: &str,
- label: &[u8],
- proof: &IdentityProof,
- ) -> Option<String> {
- let message = signed(label, address, &link.own_address());
+ /// Notes the node `proof` proves to be at `address` (as paths keep it); None if it
+ /// proves nothing.
+ fn prove(&self, address: &str, label: &[u8], proof: &IdentityProof) -> Option<String> {
+ let message = signed(label, address, &self.paths.own_address(address)?);
if !identity::verify(&proof.key, &message, &proof.signature) {
self.lock().counts.proofs_rejected += 1;
debug!(from = %address, "Ignoring an identity proof that doesn't hold");
@@ -345,22 +277,26 @@ impl Shared {
/// Sends a WHO or ME (`kind`) to `address`.
async fn send_proof(&self, kind: MessageKind, address: &str) {
- let Some((link, target)) = self.split(address) else {
+ let Some(address) = self.paths.canonical(address) else {
return;
};
+ let address = address.as_str();
let message = if kind == MessageKind::Who {
- ParsedMessage::Who(self.proof(WHO_LABEL, link, &target))
+ self.proof(WHO_LABEL, address).map(ParsedMessage::Who)
} else {
- ParsedMessage::Me(self.proof(ME_LABEL, link, &target))
+ self.proof(ME_LABEL, address).map(ParsedMessage::Me)
};
- let encoded = match link.inner.encode(&target, &message) {
+ let Some(message) = message else {
+ return;
+ };
+ let encoded = match self.paths.encode(address, &message) {
Ok(encoded) => encoded,
Err(error) => {
debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof");
return;
}
};
- match link.inner.send_encoded(&encoded).await {
+ match self.paths.send(&encoded).await {
Ok(()) => self
.traffic
.sent(address, encoded.kind, encoded.payload_len),
@@ -372,22 +308,19 @@ impl Shared {
}
}
- /// A message from `sender` on `links[index]`.
+ /// A message from `address`.
async fn receive(
self: Arc<Self>,
- index: usize,
- sender: String,
+ address: String,
message: ParsedMessage,
payload_len: usize,
handler: &HandlerFn,
) -> Result<()> {
- let link = &self.links[index];
- let address = link.address(&sender);
match message {
ParsedMessage::Who(proof) => {
self.traffic
.received(&address, MessageKind::Who, payload_len);
- if self.prove(link, &address, WHO_LABEL, &proof).is_some() {
+ if self.prove(&address, WHO_LABEL, &proof).is_some() {
let shared = Arc::clone(&self);
tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await });
}
@@ -396,7 +329,7 @@ impl Shared {
ParsedMessage::Me(proof) => {
self.traffic
.received(&address, MessageKind::Me, payload_len);
- self.prove(link, &address, ME_LABEL, &proof);
+ self.prove(&address, ME_LABEL, &proof);
Ok(())
}
message => {
@@ -425,29 +358,14 @@ impl Shared {
let pick = |addresses: Option<&Vec<String>>| {
addresses?
.iter()
- .filter(|address| self.split(address).is_some())
- .min_by_key(|address| self.preference(address))
+ .filter(|address| self.paths.usable(address))
+ .min_by_key(|address| self.paths.preference(address))
.cloned()
};
pick(book.proven.get(peer)).or_else(|| pick(book.hints.get(peer)))
}
}
-/// The first of the `results` to arrive, from tasks that run until they end.
-async fn first<T: Send + 'static>(
- tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>,
-) -> Option<T> {
- let (sender, mut receiver) = mpsc::channel(1);
- for task in tasks {
- let sender = sender.clone();
- tokio::spawn(async move {
- let _ = sender.send(task.await).await;
- });
- }
- drop(sender);
- receiver.recv().await
-}
-
#[async_trait]
impl Transport for IdentityTransport {
fn local_id(&self) -> &str {
@@ -458,28 +376,12 @@ impl Transport for IdentityTransport {
identity::is_node_id(id)
}
- /// The longest of the transports': until then, something may still be on its way.
fn delivery_limit(&self) -> Duration {
- self.shared
- .links
- .iter()
- .map(|link| link.inner.delivery_limit())
- .max()
- .unwrap_or_default()
+ self.shared.paths.delivery_limit()
}
- /// Each transport's (`Transport::local_addresses`), with its scheme.
fn local_addresses(&self) -> Vec<String> {
- self.shared
- .links
- .iter()
- .flat_map(|link| {
- link.inner
- .local_addresses()
- .into_iter()
- .map(|target| link.address(&target))
- })
- .collect()
+ self.shared.paths.local_addresses()
}
/// Where `peer` proved itself, else where it is said to be, as others can use it.
@@ -494,7 +396,7 @@ impl Transport for IdentityTransport {
};
let mut locations = Vec::new();
for address in addresses {
- for location in self.shared.locations(&address) {
+ for location in self.shared.paths.locations(&address) {
if !locations.contains(&location) {
locations.push(location);
}
@@ -509,8 +411,8 @@ impl Transport for IdentityTransport {
}
let mut known: Vec<String> = Vec::new();
for address in addresses {
- if let Some(address) = self.shared.canonical(address)
- && !self.shared.is_own(&address)
+ if let Some(address) = self.shared.paths.canonical(address)
+ && !self.shared.paths.is_own(&address)
&& !known.contains(&address)
{
known.push(address);
@@ -531,11 +433,7 @@ impl Transport for IdentityTransport {
}
fn transports(&self) -> Vec<String> {
- self.shared
- .links
- .iter()
- .map(|link| link.scheme.to_owned())
- .collect()
+ self.shared.paths.transports()
}
/// The address `encode` picks, and the transport's connection there.
@@ -547,10 +445,7 @@ impl Transport for IdentityTransport {
.proven
.get(peer)
.is_some_and(|addresses| addresses.contains(&address));
- let connection = self
- .shared
- .split(&address)
- .and_then(|(link, target)| link.inner.connection(&target));
+ let connection = self.shared.paths.connection(&address);
Some(Destination {
address,
proven,
@@ -560,28 +455,16 @@ impl Transport for IdentityTransport {
/// The identity layer's, and each transport's.
fn stats(&self) -> TransportStats {
- let mut stats = TransportStats {
+ let mut stats = self.shared.paths.stats();
+ stats.merge(TransportStats {
identity: Some(self.shared.lock().stats()),
..TransportStats::default()
- };
- for link in &self.shared.links {
- stats.merge(link.inner.stats());
- }
+ });
stats
}
- /// Each transport's, with its scheme.
fn discovered_seeds(&self) -> Vec<String> {
- self.shared
- .links
- .iter()
- .flat_map(|link| {
- link.inner
- .discovered_seeds()
- .into_iter()
- .map(|seed| link.address(&seed))
- })
- .collect()
+ self.shared.paths.discovered_seeds()
}
/// A node ID resolves to itself if this node knows where it is; anything else is
@@ -594,45 +477,8 @@ impl Transport for IdentityTransport {
}
bail!("a node ID alone doesn't say where that node is; give its address");
}
- // With a scheme of ours, only that transport resolves the rest (an address the
- // transport can't read as an identity, e.g. a libp2p multiaddr without its peer
- // ID, still resolves); without one, the first transport that can.
- let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
- let link = self
- .shared
- .links
- .iter()
- .find(|link| link.scheme == scheme)?;
- Some((link, rest))
- });
- let address = match self.shared.canonical(seed) {
- Some(address) => address,
- None => match scheme_and_rest {
- Some((link, rest)) => {
- let target = link
- .inner
- .resolve_seed(rest)
- .await
- .with_context(|| format!("{}: {seed}", link.scheme))?;
- link.address(&target)
- }
- None => {
- let mut failures = Vec::new();
- let mut resolved = None;
- for link in &self.shared.links {
- match link.inner.resolve_seed(seed).await {
- Ok(target) => {
- resolved = Some(link.address(&target));
- break;
- }
- Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
- }
- }
- resolved.ok_or_else(|| anyhow!("{}", failures.join("; ")))?
- }
- },
- };
- if self.shared.is_own(&address) {
+ let address = self.shared.paths.resolve(seed).await?;
+ if self.shared.paths.is_own(&address) {
bail!("{seed} is this node");
}
let receiver = {
@@ -667,12 +513,8 @@ impl Transport for IdentityTransport {
let Some(address) = self.shared.destination(peer) else {
bail!("no address known for {peer} on this node's transports");
};
- let Some((link, target)) = self.shared.split(&address) else {
- bail!("no transport for {address}");
- };
- let mut encoded = link.inner.encode(&target, message)?;
+ let mut encoded = self.shared.paths.encode(&address, message)?;
encoded.peer = peer.to_owned();
- encoded.address = address;
Ok(encoded)
}
@@ -696,58 +538,29 @@ impl Transport for IdentityTransport {
}
return Err(NotProven(message.peer.clone()).into());
}
- let Some((link, target)) = self.shared.split(&message.address) else {
- bail!("no transport for {}", message.address);
- };
- let mut inner = message.clone();
- inner.peer = target.clone();
- inner.address = target;
- link.inner.send_encoded(&inner).await
+ self.shared.paths.send(message).await
}
- /// Every transport's consumer; returns when the first one does.
+ /// Every transport's consumer (`Paths::start_consumer`); returns when the first
+ /// one does.
async fn start_consumer(&self, handler: MessageHandler) -> Result<()> {
let handler: Arc<HandlerFn> = Arc::from(handler);
- let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
- Vec::new();
- for (index, link) in self.shared.links.iter().enumerate() {
- let inner = Arc::clone(&link.inner);
- let shared = Arc::clone(&self.shared);
+ let shared = Arc::clone(&self.shared);
+ let receive: Arc<PathHandler> = Arc::new(move |address, message, payload_len| {
+ let shared = Arc::clone(&shared);
let handler = Arc::clone(&handler);
- consumers.push(Box::pin(async move {
- inner
- .start_consumer(Box::new(move |sender, message, payload_len| {
- let shared = Arc::clone(&shared);
- let handler = Arc::clone(&handler);
- Box::pin(async move {
- shared
- .receive(index, sender, message, payload_len, &*handler)
- .await
- })
- }))
+ Box::pin(async move {
+ shared
+ .receive(address, message, payload_len, &*handler)
.await
- }));
- }
- first(consumers)
- .await
- .unwrap_or_else(|| Err(anyhow!("no transports")))
+ })
+ });
+ self.shared.paths.start_consumer(receive).await
}
/// When the first transport exits.
async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
- let waits = self
- .shared
- .links
- .iter()
- .map(|link| {
- let inner = Arc::clone(&link.inner);
- Box::pin(async move { inner.wait_for_exit().await })
- as std::pin::Pin<Box<dyn Future<Output = _> + Send>>
- })
- .collect();
- first(waits)
- .await
- .unwrap_or_else(|| Err(std::io::Error::other("no transports")))
+ self.shared.paths.wait_for_exit().await
}
}
diff --git a/src/transport/mod.rs b/src/transport/mod.rs
index b0d0723..e46e82e 100644
--- a/src/transport/mod.rs
+++ b/src/transport/mod.rs
@@ -16,6 +16,7 @@ pub(crate) mod identity;
pub(crate) mod mycelium;
#[cfg(feature = "libp2p")]
pub(crate) mod p2p;
+pub(crate) mod paths;
#[cfg(test)]
pub(crate) mod sim;
pub(crate) mod stats;
diff --git a/src/transport/paths.rs b/src/transport/paths.rs
new file mode 100644
index 0000000..d1a4dc9
--- /dev/null
+++ b/src/transport/paths.rs
@@ -0,0 +1,364 @@
+//! Paths: one address space over any number of transports, below the identity layer
+//! (`transport::identity`). An address is `<scheme>:<target>`: the transport by its
+//! scheme (`mycelium`, `libp2p`, …), and on it a transport identity (Mycelium's public
+//! key, libp2p's peer ID) or a location (`Transport::locate`), e.g. a multiaddr.
+//! Addresses as paths keeps them are `<scheme>:<transport identity>` (`canonical`).
+//!
+//! Every transport carries the same messages (`wire`); paths routes each send to the
+//! transport of its address, hands every transport's messages to one handler under the
+//! sender's address, and puts together what the transports say about themselves
+//! (addresses, seeds they found, figures). The transports are in order of preference,
+//! as configured (`--transports`).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::{Context, Result, anyhow};
+
+use crate::transport::stats::TransportStats;
+use crate::transport::{Connection, EncodedMessage, ParsedMessage, Transport};
+
+/// Called with the sender's address, the message and its `payload_len`.
+pub(crate) type PathHandler = dyn Fn(String, ParsedMessage, usize) -> std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>
+ + Send
+ + Sync
+ + 'static;
+
+/// One of the node's transports.
+struct Link {
+ scheme: &'static str,
+ inner: Arc<dyn Transport>,
+}
+
+impl Link {
+ fn address(&self, target: &str) -> String {
+ format!("{}:{target}", self.scheme)
+ }
+
+ fn own_address(&self) -> String {
+ self.address(self.inner.local_id())
+ }
+}
+
+/// The scheme of `address`.
+pub(crate) fn scheme_of(address: &str) -> &str {
+ address.split_once(':').map_or("", |(scheme, _)| scheme)
+}
+
+pub(crate) struct Paths {
+ /// In order of preference.
+ links: Vec<Link>,
+}
+
+impl Paths {
+ /// Over `transports`, by scheme, in order of preference.
+ pub(crate) fn new(transports: Vec<(&'static str, Arc<dyn Transport>)>) -> Self {
+ Self {
+ links: transports
+ .into_iter()
+ .map(|(scheme, inner)| Link { scheme, inner })
+ .collect(),
+ }
+ }
+
+ /// The transport `address` is on, and the transport identity at it
+ /// (`Transport::locate`, which notes where that is if the address says); None if no
+ /// transport of ours has its scheme, or it isn't one of that transport's.
+ fn split(&self, address: &str) -> Option<(&Link, String)> {
+ let (scheme, target) = address.split_once(':')?;
+ let link = self.links.iter().find(|link| link.scheme == scheme)?;
+ let target = link.inner.locate(target)?;
+ Some((link, target))
+ }
+
+ /// Whether one of our transports can use `address`.
+ pub(crate) fn usable(&self, address: &str) -> bool {
+ self.split(address).is_some()
+ }
+
+ /// `address` as paths keep it: `<scheme>:<transport identity>`.
+ pub(crate) fn canonical(&self, address: &str) -> Option<String> {
+ self.split(address)
+ .map(|(link, target)| link.address(&target))
+ }
+
+ /// Where the node at `address` (as paths keep it) is, as others can use it
+ /// (`Transport::locations`).
+ pub(crate) fn locations(&self, address: &str) -> Vec<String> {
+ let Some((link, target)) = self.split(address) else {
+ return Vec::new();
+ };
+ link.inner
+ .locations(&target)
+ .iter()
+ .map(|location| link.address(location))
+ .collect()
+ }
+
+ /// Where `address`'s transport is in the order of preference (none of ours: last).
+ pub(crate) fn preference(&self, address: &str) -> usize {
+ let scheme = scheme_of(address);
+ self.links
+ .iter()
+ .position(|link| link.scheme == scheme)
+ .unwrap_or(usize::MAX)
+ }
+
+ /// Whether `address` is one of ours.
+ pub(crate) fn is_own(&self, address: &str) -> bool {
+ self.links.iter().any(|link| link.own_address() == address)
+ }
+
+ /// Our address on the transport of `address`, as `address` names it there.
+ pub(crate) fn own_address(&self, address: &str) -> Option<String> {
+ let scheme = scheme_of(address);
+ self.links
+ .iter()
+ .find(|link| link.scheme == scheme)
+ .map(Link::own_address)
+ }
+
+ /// The longest of the transports': until then, something may still be on its way.
+ pub(crate) fn delivery_limit(&self) -> Duration {
+ self.links
+ .iter()
+ .map(|link| link.inner.delivery_limit())
+ .max()
+ .unwrap_or_default()
+ }
+
+ /// Each transport's (`Transport::local_addresses`), with its scheme.
+ pub(crate) fn local_addresses(&self) -> Vec<String> {
+ self.links
+ .iter()
+ .flat_map(|link| {
+ link.inner
+ .local_addresses()
+ .into_iter()
+ .map(|target| link.address(&target))
+ })
+ .collect()
+ }
+
+ /// The schemes of the transports, in order of preference.
+ pub(crate) fn transports(&self) -> Vec<String> {
+ self.links
+ .iter()
+ .map(|link| link.scheme.to_owned())
+ .collect()
+ }
+
+ /// Each transport's, with its scheme.
+ pub(crate) fn discovered_seeds(&self) -> Vec<String> {
+ self.links
+ .iter()
+ .flat_map(|link| {
+ link.inner
+ .discovered_seeds()
+ .into_iter()
+ .map(|seed| link.address(&seed))
+ })
+ .collect()
+ }
+
+ /// The address of the node a seed leads to, as paths keep it: an address of ours
+ /// (`<scheme>:...`) whose transport reads it at once; with a scheme of ours, what
+ /// that transport resolves the rest to (an address the transport can't read as an
+ /// identity, e.g. a libp2p multiaddr without its peer ID, still resolves); without
+ /// one, what the first transport that can resolves it to.
+ pub(crate) async fn resolve(&self, seed: &str) -> Result<String> {
+ if let Some(address) = self.canonical(seed) {
+ return Ok(address);
+ }
+ let scheme_and_rest = seed.split_once(':').and_then(|(scheme, rest)| {
+ let link = self.links.iter().find(|link| link.scheme == scheme)?;
+ Some((link, rest))
+ });
+ if let Some((link, rest)) = scheme_and_rest {
+ let target = link
+ .inner
+ .resolve_seed(rest)
+ .await
+ .with_context(|| format!("{}: {seed}", link.scheme))?;
+ return Ok(link.address(&target));
+ }
+ let mut failures = Vec::new();
+ for link in &self.links {
+ match link.inner.resolve_seed(seed).await {
+ Ok(target) => return Ok(link.address(&target)),
+ Err(error) => failures.push(format!("{}: {error:#}", link.scheme)),
+ }
+ }
+ Err(anyhow!("{}", failures.join("; ")))
+ }
+
+ /// `message` encoded for `address`: `peer` and `address` of the result are
+ /// `address`.
+ pub(crate) fn encode(&self, address: &str, message: &ParsedMessage) -> Result<EncodedMessage> {
+ let (link, target) = self
+ .split(address)
+ .with_context(|| format!("no transport for {address}"))?;
+ let mut encoded = link.inner.encode(&target, message)?;
+ encoded.peer = address.to_owned();
+ encoded.address = address.to_owned();
+ Ok(encoded)
+ }
+
+ /// Sends `message` to its `address`.
+ pub(crate) async fn send(&self, message: &EncodedMessage) -> Result<()> {
+ let (link, target) = self
+ .split(&message.address)
+ .with_context(|| format!("no transport for {}", message.address))?;
+ let mut inner = message.clone();
+ inner.peer = target.clone();
+ inner.address = target;
+ link.inner.send_encoded(&inner).await
+ }
+
+ /// How the transport of `address` is connected there (`Transport::connection`).
+ pub(crate) fn connection(&self, address: &str) -> Option<Connection> {
+ let (link, target) = self.split(address)?;
+ link.inner.connection(&target)
+ }
+
+ /// Every transport's figures together.
+ pub(crate) fn stats(&self) -> TransportStats {
+ let mut stats = TransportStats::default();
+ for link in &self.links {
+ stats.merge(link.inner.stats());
+ }
+ stats
+ }
+
+ /// Every transport's consumer, each message handed to `handler` under the sender's
+ /// address; returns when the first one does.
+ pub(crate) async fn start_consumer(&self, handler: Arc<PathHandler>) -> Result<()> {
+ let mut consumers: Vec<std::pin::Pin<Box<dyn Future<Output = Result<()>> + Send>>> =
+ Vec::new();
+ for link in &self.links {
+ let inner = Arc::clone(&link.inner);
+ let scheme = link.scheme;
+ let handler = Arc::clone(&handler);
+ consumers.push(Box::pin(async move {
+ inner
+ .start_consumer(Box::new(move |sender, message, payload_len| {
+ handler(format!("{scheme}:{sender}"), message, payload_len)
+ }))
+ .await
+ }));
+ }
+ first(consumers)
+ .await
+ .unwrap_or_else(|| Err(anyhow!("no transports")))
+ }
+
+ /// When the first transport exits.
+ pub(crate) async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> {
+ let waits = self
+ .links
+ .iter()
+ .map(|link| {
+ let inner = Arc::clone(&link.inner);
+ Box::pin(async move { inner.wait_for_exit().await })
+ as std::pin::Pin<Box<dyn Future<Output = _> + Send>>
+ })
+ .collect();
+ first(waits)
+ .await
+ .unwrap_or_else(|| Err(std::io::Error::other("no transports")))
+ }
+}
+
+/// The first of the `results` to arrive, from tasks that run until they end.
+async fn first<T: Send + 'static>(
+ tasks: Vec<std::pin::Pin<Box<dyn Future<Output = T> + Send>>>,
+) -> Option<T> {
+ let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
+ for task in tasks {
+ let sender = sender.clone();
+ tokio::spawn(async move {
+ let _ = sender.send(task.await).await;
+ });
+ }
+ drop(sender);
+ receiver.recv().await
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::document::replica::Epoch;
+ use crate::transport::sim::SimNet;
+ use crate::transport::{HeadsSummary, Status};
+
+ /// "a" on two sim nets, as `sim` (preferred) and `other`; "b" on both too.
+ fn two_nets() -> (Paths, Paths) {
+ let (first, second) = (SimNet::new(1), SimNet::new(2));
+ let on_both = |id: &str| {
+ Paths::new(vec![
+ ("sim", first.transport(id) as Arc<dyn Transport>),
+ ("other", second.transport(id) as Arc<dyn Transport>),
+ ])
+ };
+ (on_both("a"), on_both("b"))
+ }
+
+ #[test]
+ fn addresses_by_scheme() {
+ let (a, _b) = two_nets();
+ assert_eq!(a.local_addresses(), ["sim:a", "other:a"]);
+ assert_eq!(a.transports(), ["sim", "other"]);
+ assert_eq!(a.canonical("other:b").as_deref(), Some("other:b"));
+ assert_eq!(a.canonical("pigeon:b"), None);
+ assert_eq!(a.canonical("b"), None); // No scheme.
+ assert!(a.preference("sim:b") < a.preference("other:b"));
+ assert_eq!(a.preference("pigeon:b"), usize::MAX);
+ assert!(a.is_own("other:a") && !a.is_own("other:b"));
+ assert_eq!(a.own_address("other:b").as_deref(), Some("other:a"));
+ assert_eq!(a.own_address("pigeon:b"), None);
+ }
+
+ #[tokio::test]
+ async fn seeds_resolve_to_addresses() {
+ let (a, _b) = two_nets();
+ assert_eq!(a.resolve("other:b").await.unwrap(), "other:b");
+ // Without a scheme: the first transport that resolves it.
+ assert_eq!(a.resolve("b").await.unwrap(), "sim:b");
+ assert!(a.resolve("nobody").await.is_err());
+ }
+
+ /// A message sent to an address arrives under the sender's address on that
+ /// transport.
+ #[tokio::test]
+ async fn messages_arrive_under_the_senders_address() {
+ let (a, b) = two_nets();
+ let (sender, mut received) = tokio::sync::mpsc::unbounded_channel();
+ let handler: Arc<PathHandler> = Arc::new(move |from, message, _| {
+ let sender = sender.clone();
+ Box::pin(async move {
+ let _ = sender.send((from, message.kind()));
+ Ok(())
+ })
+ });
+ tokio::spawn(async move { b.start_consumer(handler).await });
+ let status = ParsedMessage::Status(Status {
+ heads: HeadsSummary::of(Vec::new()),
+ epoch: Epoch::GENESIS,
+ epoch_seq: 0,
+ ready: false,
+ syncs_with_you: false,
+ });
+ for to in ["other:b", "sim:b"] {
+ let encoded = a.encode(to, &status).unwrap();
+ assert_eq!(encoded.address, to);
+ a.send(&encoded).await.unwrap();
+ }
+ let mut from: Vec<String> = Vec::new();
+ for _ in 0..2 {
+ from.push(received.recv().await.unwrap().0);
+ }
+ from.sort();
+ assert_eq!(from, ["other:a", "sim:a"]);
+ assert!(a.encode("pigeon:b", &status).is_err());
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-28 07:55:19 +0000