asimov_protocol/
node.rs

1// This is free and unencumbered software released into the public domain.
2
3use crate::{
4    BindError, ConnectError, DefaultPreset, GOSSIP_ALPN, GossipProtocol, NODE_ALPN, NodeProtocol,
5    PeerConnect, PeerConnection, PingError, StartError, SubscribeError, TerminateError, Topic,
6    TopicSubscription, node_state::*,
7};
8use alloc::vec::Vec;
9use asimov_id::PublicKey;
10use core::{result::Result, time::Duration};
11use iroh::{Endpoint, EndpointAddr, EndpointId, endpoint::EndpointClosed, protocol::Router};
12
13#[derive(Debug)]
14pub struct Node<State = Building>(State);
15
16impl Default for Node {
17    fn default() -> Self {
18        Self(Building {
19            endpoint: Endpoint::builder(DefaultPreset),
20        })
21    }
22}
23
24impl Node<Building> {
25    pub async fn bind(self) -> Result<Node<Bound>, BindError> {
26        Ok(Node(Bound {
27            endpoint: self.0.endpoint.bind().await?,
28        }))
29    }
30}
31
32impl Node<Bound> {
33    pub fn public_key(&self) -> PublicKey {
34        self.endpoint().id().into()
35    }
36
37    pub fn endpoint_addr(&self) -> EndpointAddr {
38        self.endpoint().addr()
39    }
40
41    pub fn endpoint(&self) -> &Endpoint {
42        &self.0.endpoint
43    }
44
45    pub async fn start(self) -> Result<Node<Running>, StartError> {
46        let endpoint = self.0.endpoint;
47        endpoint.online().await;
48        let node = NodeProtocol::new();
49        let gossip = GossipProtocol::new(endpoint.clone());
50        let router = Router::builder(endpoint)
51            .accept(NODE_ALPN, node.clone())
52            .accept(GOSSIP_ALPN, gossip.0.clone())
53            .spawn();
54        Ok(Node(Running {
55            router,
56            node,
57            gossip,
58            peers: Vec::new(),
59        }))
60    }
61}
62
63impl Node<Running> {
64    pub fn public_key(&self) -> PublicKey {
65        self.endpoint().id().into()
66    }
67
68    pub fn endpoint_addr(&self) -> EndpointAddr {
69        self.endpoint().addr()
70    }
71
72    pub fn endpoint(&self) -> &Endpoint {
73        self.0.router.endpoint()
74    }
75
76    pub fn is_closed(&self) -> bool {
77        self.endpoint().is_closed()
78    }
79
80    pub fn closed(&self) -> EndpointClosed {
81        self.endpoint().closed()
82    }
83
84    pub async fn terminate(self) -> Result<Node<Terminating>, TerminateError> {
85        let router = self.0.router;
86        router.shutdown().await?;
87        Ok(Node(Terminating { router }))
88    }
89
90    pub async fn online(&self) {
91        self.endpoint().online().await
92    }
93
94    pub fn add_peer(&mut self, endpoint: impl Into<EndpointId>) {
95        self.0.peers.push(endpoint.into());
96    }
97
98    pub async fn ping(&self, peer_addr: impl Into<EndpointAddr>) -> Result<Duration, PingError> {
99        let mut connection = self.connect(peer_addr).await?;
100        let rtt = connection.ping().await?;
101        Ok(rtt)
102    }
103
104    pub async fn connect(
105        &self,
106        peer_addr: impl Into<EndpointAddr>,
107    ) -> Result<PeerConnection, ConnectError> {
108        let endpoint = self.0.router.endpoint();
109
110        let state: PeerConnect = endpoint.connect(peer_addr, NODE_ALPN).await?.into();
111        let state = state.send_hello().await?;
112        let state = state.recv_hello().await?;
113
114        Ok(state.into_connection())
115    }
116
117    pub async fn subscribe(
118        &self,
119        topic: impl Into<Topic>,
120    ) -> Result<TopicSubscription, SubscribeError> {
121        let topic_id = topic.into().id();
122        let gossip = &self.0.gossip;
123        Ok(gossip
124            .0
125            .subscribe(topic_id, self.0.peers.clone())
126            .await?
127            .into())
128    }
129
130    pub async fn subscribe_and_join(
131        &self,
132        topic: impl Into<Topic>,
133    ) -> Result<TopicSubscription, SubscribeError> {
134        let topic_id = topic.into().id();
135        let gossip = &self.0.gossip;
136        Ok(gossip
137            .0
138            .subscribe_and_join(topic_id, self.0.peers.clone())
139            .await?
140            .into())
141    }
142}
143
144impl Node<Terminating> {}
asimov_protocol/node.rs

asimov_protocol/
node.rs
