use std::io;
use std::sync::{Arc, Mutex, Weak};
use rns_net::{
AnnouncedIdentity, Callbacks, DestHash, IdentityHash, LinkId, NodeConfig as RnsNodeConfig,
PacketHash, RnsNode, TeardownReason,
};
use tracing::warn;
use crate::browser::NomadBrowser;
use crate::directory::NomadDirectory;
use crate::rrc::RRCManager;
use crate::rrc_worker::RrcWorker;
#[derive(Clone)]
pub struct NomadClientCallbacks {
inner: Arc<ClientCallbackState>,
}
struct ClientCallbackState {
browser: Arc<NomadBrowser>,
directory: Arc<Mutex<NomadDirectory>>,
node: Mutex<Option<Weak<RnsNode>>>,
rrc_worker: Mutex<Option<RrcWorker>>,
}
impl NomadClientCallbacks {
pub fn new(browser: Arc<NomadBrowser>, directory: Arc<Mutex<NomadDirectory>>) -> Self {
Self {
inner: Arc::new(ClientCallbackState {
browser,
directory,
node: Mutex::new(None),
rrc_worker: Mutex::new(None),
}),
}
}
pub fn attach_node(&self, node: &Arc<RnsNode>) {
*self.inner.node.lock().unwrap_or_else(|e| e.into_inner()) = Some(Arc::downgrade(node));
}
pub fn detach_node(&self) {
*self.inner.node.lock().unwrap_or_else(|e| e.into_inner()) = None;
}
pub fn attach_rrc_worker(&self, worker: RrcWorker) {
*self
.inner
.rrc_worker
.lock()
.unwrap_or_else(|e| e.into_inner()) = Some(worker);
}
pub fn detach_rrc_worker(&self) {
*self
.inner
.rrc_worker
.lock()
.unwrap_or_else(|e| e.into_inner()) = None;
}
fn node(&self) -> Option<Arc<RnsNode>> {
self.inner
.node
.lock()
.unwrap_or_else(|e| e.into_inner())
.as_ref()
.and_then(Weak::upgrade)
}
fn rrc_worker(&self) -> Option<RrcWorker> {
self.inner
.rrc_worker
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
}
impl Callbacks for NomadClientCallbacks {
fn on_announce(&mut self, announced: AnnouncedIdentity) {
self.inner
.directory
.lock()
.unwrap_or_else(|e| e.into_inner())
.handle_announce(&announced);
}
fn on_path_updated(&mut self, _dest_hash: DestHash, _hops: u8) {}
fn on_local_delivery(&mut self, _dest_hash: DestHash, _raw: Vec<u8>, _packet_hash: PacketHash) {
}
fn on_link_established(
&mut self,
link_id: LinkId,
dest_hash: DestHash,
rtt: f64,
is_initiator: bool,
) {
if self.inner.browser.owns_link(&link_id) {
let node = self.node();
if let Err(error) = self.inner.browser.handle_link_established_with_node(
node.as_ref(),
link_id,
dest_hash.0,
) {
warn!("NomadClient: browser link setup failed: {error}");
}
}
if is_initiator {
if let Some(worker) = self.rrc_worker() {
worker.on_link_established(link_id.0, dest_hash.0, rtt);
}
}
}
fn on_link_closed(&mut self, link_id: LinkId, reason: Option<TeardownReason>) {
if self.inner.browser.owns_link(&link_id) {
self.inner
.browser
.handle_link_closed(link_id, reason.map(|value| format!("{value:?}")));
}
if let Some(worker) = self.rrc_worker() {
worker.on_link_closed(link_id.0);
}
}
fn on_remote_identified(
&mut self,
link_id: LinkId,
identity_hash: IdentityHash,
_public_key: [u8; 64],
) {
if let Some(worker) = self.rrc_worker() {
worker.on_remote_identified(link_id.0, identity_hash.0);
}
}
fn on_resource_received(&mut self, link_id: LinkId, data: Vec<u8>, _metadata: Option<Vec<u8>>) {
if let Some(worker) = self.rrc_worker() {
worker.on_resource_received(link_id.0, data);
}
}
fn on_resource_failed(&mut self, link_id: LinkId, _error: String) {
if self.inner.browser.owns_link(&link_id) {
self.inner.browser.reset_transfer_progress(&link_id);
}
if let Some(worker) = self.rrc_worker() {
worker.on_resource_failed(link_id.0);
}
}
fn on_resource_progress(&mut self, link_id: LinkId, received: usize, total: usize) {
if !self.inner.browser.owns_link(&link_id) {
return;
}
let progress = if total == 0 {
0.0
} else {
received as f64 / total as f64
};
self.inner.browser.handle_transfer_progress(
link_id,
progress,
Some(received as u64),
Some(total as u64),
);
}
fn on_resource_accept_query(
&mut self,
link_id: LinkId,
_resource_hash: Vec<u8>,
transfer_size: u64,
_has_metadata: bool,
) -> bool {
self.rrc_worker()
.map(|worker| worker.on_resource_accept_query(link_id.0, transfer_size))
.unwrap_or(false)
}
fn on_link_data(&mut self, link_id: LinkId, _context: u8, data: Vec<u8>) {
if let Some(worker) = self.rrc_worker() {
worker.on_link_data(link_id.0, data);
}
}
fn on_response(&mut self, link_id: LinkId, request_id: [u8; 16], data: Vec<u8>) {
if self.inner.browser.owns_link(&link_id) {
self.inner
.browser
.handle_response(link_id, request_id, data);
}
}
}
pub struct NomadClient {
node: Option<Arc<RnsNode>>,
browser: Arc<NomadBrowser>,
directory: Arc<Mutex<NomadDirectory>>,
callbacks: NomadClientCallbacks,
rrc_worker: Option<RrcWorker>,
}
impl NomadClient {
pub fn start(config: RnsNodeConfig) -> io::Result<Self> {
Self::start_with_state(
config,
Arc::new(NomadBrowser::new()),
Arc::new(Mutex::new(NomadDirectory::new())),
)
}
pub fn start_with_state(
config: RnsNodeConfig,
browser: Arc<NomadBrowser>,
directory: Arc<Mutex<NomadDirectory>>,
) -> io::Result<Self> {
let callbacks = NomadClientCallbacks::new(browser.clone(), directory.clone());
let node = Arc::new(RnsNode::start(config, Box::new(callbacks.clone()))?);
callbacks.attach_node(&node);
Ok(Self {
node: Some(node),
browser,
directory,
callbacks,
rrc_worker: None,
})
}
pub fn node(&self) -> &Arc<RnsNode> {
self.node.as_ref().expect("client node is unavailable")
}
pub fn browser(&self) -> &Arc<NomadBrowser> {
&self.browser
}
pub fn directory(&self) -> &Arc<Mutex<NomadDirectory>> {
&self.directory
}
pub fn callbacks(&self) -> &NomadClientCallbacks {
&self.callbacks
}
pub fn enable_rrc(
&mut self,
manager: RRCManager,
identity_prv: [u8; 64],
identity_hash: [u8; 16],
) -> io::Result<Arc<Mutex<RRCManager>>> {
if self.rrc_worker.is_some() {
return Err(io::Error::new(
io::ErrorKind::AlreadyExists,
"RRC worker is already enabled",
));
}
if tokio::runtime::Handle::try_current().is_err() {
return Err(io::Error::other("RRC worker requires a Tokio runtime"));
}
let worker = RrcWorker::new(manager, self.node().clone(), identity_prv, identity_hash);
let manager = worker.manager();
self.callbacks.attach_rrc_worker(worker.clone());
self.rrc_worker = Some(worker);
Ok(manager)
}
pub fn rrc_manager(&self) -> Option<Arc<Mutex<RRCManager>>> {
self.rrc_worker.as_ref().map(RrcWorker::manager)
}
pub async fn disable_rrc(&mut self) {
self.callbacks.detach_rrc_worker();
if let Some(worker) = self.rrc_worker.take() {
worker.shutdown();
worker.wait_stopped().await;
}
}
pub async fn shutdown(mut self) -> io::Result<()> {
self.disable_rrc().await;
self.callbacks.detach_node();
let node = self.node.take().expect("client node is unavailable");
let node = Arc::try_unwrap(node).map_err(|node| {
io::Error::other(format!(
"cannot shut down RNS node while {} shared references remain",
Arc::strong_count(&node) - 1
))
})?;
node.shutdown();
Ok(())
}
}
impl Drop for NomadClient {
fn drop(&mut self) {
self.callbacks.detach_rrc_worker();
self.callbacks.detach_node();
if let Some(worker) = self.rrc_worker.take() {
worker.shutdown();
}
}
}
#[cfg(test)]
mod tests {
use rns_core::destination::destination_hash;
use rns_net::{InterfaceId, TeardownReason};
use super::*;
use crate::types::BrowseEvent;
fn callbacks() -> (
NomadClientCallbacks,
Arc<NomadBrowser>,
Arc<Mutex<NomadDirectory>>,
) {
let browser = Arc::new(NomadBrowser::new());
let directory = Arc::new(Mutex::new(NomadDirectory::new()));
(
NomadClientCallbacks::new(browser.clone(), directory.clone()),
browser,
directory,
)
}
#[test]
fn routes_nomadnet_announces_to_directory() {
let (mut callbacks, _, directory) = callbacks();
let identity_hash = [0xA5; 16];
let announced = AnnouncedIdentity {
dest_hash: DestHash(destination_hash(
"nomadnetwork",
&["node"],
Some(&identity_hash),
)),
identity_hash: IdentityHash(identity_hash),
public_key: [0; 64],
app_data: Some(b"Test Node".to_vec()),
hops: 2,
received_at: 1000.0,
receiving_interface: InterfaceId(0),
};
callbacks.on_announce(announced);
let directory = directory.lock().unwrap_or_else(|e| e.into_inner());
assert_eq!(directory.len(), 1);
assert_eq!(
directory.known_nodes()[0].node_name.as_deref(),
Some("Test Node")
);
}
#[test]
fn routes_owned_link_events_to_browser() {
let (mut callbacks, browser, _) = callbacks();
let mut events = browser.events();
let link_id = LinkId([0x12; 16]);
let dest_hash = [0x34; 16];
browser.handle_link_established(link_id, dest_hash);
assert!(matches!(
events.try_recv().unwrap(),
BrowseEvent::LinkEstablished { .. }
));
callbacks.on_resource_progress(link_id, 25, 100);
match events.try_recv().unwrap() {
BrowseEvent::TransferProgress {
progress,
data_size,
transfer_size,
..
} => {
assert_eq!(progress, 0.25);
assert_eq!(data_size, Some(25));
assert_eq!(transfer_size, Some(100));
}
event => panic!("expected transfer progress, got {event:?}"),
}
callbacks.on_link_closed(link_id, Some(TeardownReason::Timeout));
assert!(matches!(
events.try_recv().unwrap(),
BrowseEvent::LinkClosed { .. }
));
assert!(!browser.owns_link(&link_id));
}
#[test]
fn ignores_unowned_browser_links() {
let (mut callbacks, browser, _) = callbacks();
let mut events = browser.events();
let link_id = LinkId([0x56; 16]);
callbacks.on_link_established(link_id, DestHash([0x78; 16]), 0.1, true);
callbacks.on_resource_progress(link_id, 1, 2);
assert!(!browser.owns_link(&link_id));
assert!(events.try_recv().is_err());
}
#[test]
fn rejects_resources_without_rrc_expectations() {
let (mut callbacks, _, _) = callbacks();
assert!(!callbacks.on_resource_accept_query(LinkId([0x9A; 16]), vec![1, 2, 3], 100, false,));
}
}