use super::NetworkBackend;
use crate::config::types::OutboundLinkConfig;
use crate::link_carrier::{Carrier, RelayCarrier, read_frame, write_frame};
use crate::link_dial::{LinkDialOutcome, LinkDialer};
use crate::link_open::carriers::{DirectPath, StreamPath, named};
use crate::link_open::{Backend, OpenError, Path, Stop};
use liminal::protocol::Frame;
use std::sync::Arc;
impl Backend for NetworkBackend {
fn interrupt_handle(&self) -> Option<Stop> {
let network = Arc::clone(&self.network);
Some(Arc::new(move || network.stop()))
}
fn wire(
&mut self,
config: &OutboundLinkConfig,
address: &str,
) -> Result<Box<dyn Path>, OpenError> {
self.require_config(config)?;
self.network.require_running()?;
let mut entry = config.clone();
entry.peer_address = Some(address.to_owned());
match LinkDialer::new(&self.network.host, &self.network.machine)
.with_network(&self.network)
.open(&entry)
.map_err(named)?
{
LinkDialOutcome::Opened(link) => {
let interrupt = self
.network
.retain(link.carrier.interrupt_handle().map_err(named)?)?;
Ok(Box::new(StreamPath::wire(*link).with_interrupt(interrupt)))
}
LinkDialOutcome::Held { grant, draft } => {
Err(OpenError(format!("link_held:{grant}:{draft}")))
}
LinkDialOutcome::Refused { reason } => Err(OpenError(format!("link_refused:{reason}"))),
}
}
fn lookup(&mut self, config: &OutboundLinkConfig) -> Result<Vec<String>, OpenError> {
self.require_config(config)?;
self.network.require_running()?;
let (mut broker, connections) = self.network.broker()?;
let result = broker
.call(
"net.broker.lookup",
&serde_json::json!({"device": hex::encode(self.peer.ed25519_public())}),
)
.map_err(named);
drop(connections);
let value = result?;
let addresses: Vec<String> = serde_json::from_value(
value
.get("addresses")
.cloned()
.ok_or_else(|| OpenError("invalid_broker_answer".to_owned()))?,
)
.map_err(|_| OpenError("invalid_broker_answer".to_owned()))?;
if addresses.len() > 64
|| addresses
.iter()
.any(|address| address.is_empty() || address.len() > 2048)
{
return Err(OpenError("invalid_broker_addresses".to_owned()));
}
Ok(addresses)
}
fn punch(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
self.require_config(config)?;
let index = {
let mut direct = self.network.direct.lock().map_err(named)?;
let index = direct.next;
direct.next = index
.checked_add(1)
.ok_or_else(|| OpenError("direct_indices_exhausted".to_owned()))?;
index
};
let peer_name = hex::encode(self.peer.ed25519_public());
self.network
.pending
.lock()
.map_err(named)?
.insert(peer_name.clone());
let result = self.network.with_puncher(|puncher| {
puncher.prepare_open().map_err(named)?;
let handshake_guard = self.network.handshakes.lock().map_err(named)?;
let link = puncher.punch(self.peer, index).map_err(named)?;
drop(handshake_guard);
let mut path =
DirectPath::new(puncher.shared_carrier().map_err(named)?, link.local_index)?;
let pass =
crate::link_pass::MachinePass::configured(&self.network.machine).map_err(named)?;
let token = pass.for_open().map_err(named)?;
let deadline = crate::link_open::Deadline::after(std::time::Duration::from_millis(
self.network.choice.crossing_deadline_ms,
))?;
path.admit(
&self.network.host,
&self.network.machine.agent_id,
self.peer,
config,
&token,
deadline.expires(),
)?;
Ok(Box::new(path) as Box<dyn Path>)
});
self.network
.pending
.lock()
.map_err(named)?
.remove(&peer_name);
result
}
fn relay(&mut self, config: &OutboundLinkConfig) -> Result<Box<dyn Path>, OpenError> {
self.require_config(config)?;
self.network.require_running()?;
let choice = &self.network.choice;
let holder = choice
.relay_holder
.as_deref()
.ok_or_else(|| OpenError("relay_not_configured".to_owned()))?;
let relay = self
.network
.host
.recorded_machine(holder)
.ok_or_else(|| OpenError("relay_has_no_recorded_machine".to_owned()))?;
let address = choice
.relay_address
.as_deref()
.ok_or_else(|| OpenError("relay_address_not_configured".to_owned()))?;
let (mut socket, interrupt) = self.network.connect(address)?;
write_frame(
&mut socket,
&Frame::ConversationOpen {
flags: 0,
stream_id: 1,
conversation_id: 1,
subject: format!("net.relay.{}", hex::encode(self.peer.ed25519_public())),
},
)
.map_err(named)?;
write_frame(&mut socket, &Frame::Ping { flags: 0 }).map_err(named)?;
if !matches!(
read_frame(&mut socket).map_err(named)?,
Some(Frame::Pong { .. })
) {
return Err(OpenError("relay_stream_not_marked".to_owned()));
}
let carrier = RelayCarrier::new(socket, 1, relay).map_err(named)?;
let link = crate::link_relay::open(carrier, self.network.host.local(), self.peer)
.map_err(named)?;
Ok(Box::new(
StreamPath::relay(link, hex::encode(relay.ed25519_public())).with_interrupt(interrupt),
))
}
fn repunch(
&mut self,
config: &OutboundLinkConfig,
path: &mut dyn Path,
) -> Result<(), OpenError> {
self.require_config(config)?;
let index = path
.direct_index()
.ok_or_else(|| OpenError("cannot_repunch_a_stream".to_owned()))?;
self.network
.with_puncher(|puncher| puncher.refresh(self.peer, index).map_err(named))
}
fn topology(
&mut self,
carrier: Option<Carrier>,
relay: Option<String>,
refusal: Option<&str>,
) -> Result<(), OpenError> {
use crate::console::records::{Carrier as ObservedCarrier, Link, LinkCounters, LinkState};
if refusal.is_some() {
self.refused = self
.refused
.checked_add(1)
.ok_or_else(|| OpenError("carrier_refusal_count_exhausted".to_owned()))?;
}
let peer = hex::encode(self.peer.ed25519_public());
let local = self.services.console().device();
let carrier = match carrier {
Some(Carrier::Wire) => Some(ObservedCarrier::Wire),
Some(Carrier::Direct) => Some(ObservedCarrier::Direct),
Some(Carrier::Relay) => {
let device = relay.or_else(|| {
self.network
.choice
.relay_holder
.as_deref()
.and_then(|holder| self.network.host.recorded_machine(holder))
.map(|machine| hex::encode(machine.ed25519_public()))
});
if device.is_none() && refusal.is_none() {
return Err(OpenError("relay_topology_has_no_device".to_owned()));
}
device.map(ObservedCarrier::Relay)
}
None => None,
};
self.services
.console()
.emit(crate::console::TopologyChange::Link {
link_id: Link::to_peer(local, &peer).id,
peer,
carrier,
state: if refusal.is_some() {
LinkState::Down
} else {
LinkState::Up
},
reason: refusal.map(str::to_owned),
counters: LinkCounters {
refused: Some(self.refused),
..LinkCounters::default()
},
})
.map_err(named)
}
}