|
|
|
@ -63,7 +63,7 @@ pub static BROKER: Lazy<Arc<RwLock<Broker>>> = Lazy::new(|| Arc::new(RwLock::new |
|
|
|
|
|
|
|
|
|
|
|
pub struct Broker { |
|
|
|
pub struct Broker { |
|
|
|
direct_connections: HashMap<IP, DirectConnection>, |
|
|
|
direct_connections: HashMap<IP, DirectConnection>, |
|
|
|
peers: HashMap<DirectPeerId, BrokerPeerInfo>, |
|
|
|
peers: HashMap<X25519PubKey, BrokerPeerInfo>, |
|
|
|
/// (local,remote) -> ConnectionBase
|
|
|
|
/// (local,remote) -> ConnectionBase
|
|
|
|
anonymous_connections: HashMap<(BindAddress, BindAddress), ConnectionBase>, |
|
|
|
anonymous_connections: HashMap<(BindAddress, BindAddress), ConnectionBase>, |
|
|
|
#[cfg(not(target_arch = "wasm32"))] |
|
|
|
#[cfg(not(target_arch = "wasm32"))] |
|
|
|
@ -248,7 +248,7 @@ impl Broker { |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
pub fn reconnecting(&mut self, peer_id: &DirectPeerId) { |
|
|
|
pub fn reconnecting(&mut self, peer_id: &DirectPeerId) { |
|
|
|
let peerinfo = self.peers.get_mut(peer_id); |
|
|
|
let peerinfo = self.peers.get_mut(&peer_id.to_dh_slice()); |
|
|
|
match peerinfo { |
|
|
|
match peerinfo { |
|
|
|
Some(info) => match &info.connected { |
|
|
|
Some(info) => match &info.connected { |
|
|
|
PeerConnection::NONE => {} |
|
|
|
PeerConnection::NONE => {} |
|
|
|
@ -264,7 +264,7 @@ impl Broker { |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
pub fn remove_peer_id(&mut self, peer_id: &DirectPeerId) { |
|
|
|
pub fn remove_peer_id(&mut self, peer_id: &DirectPeerId) { |
|
|
|
let removed = self.peers.remove(peer_id); |
|
|
|
let removed = self.peers.remove(&peer_id.to_dh_slice()); |
|
|
|
match removed { |
|
|
|
match removed { |
|
|
|
Some(info) => match info.connected { |
|
|
|
Some(info) => match info.connected { |
|
|
|
PeerConnection::NONE => {} |
|
|
|
PeerConnection::NONE => {} |
|
|
|
@ -285,6 +285,9 @@ impl Broker { |
|
|
|
let removed = self |
|
|
|
let removed = self |
|
|
|
.anonymous_connections |
|
|
|
.anonymous_connections |
|
|
|
.remove(&(local_bind_address, remote_bind_address)); |
|
|
|
.remove(&(local_bind_address, remote_bind_address)); |
|
|
|
|
|
|
|
if removed.is_some() { |
|
|
|
|
|
|
|
removed.unwrap().release_shutdown(); |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
pub fn test(&self) -> u32 { |
|
|
|
pub fn test(&self) -> u32 { |
|
|
|
@ -363,7 +366,7 @@ impl Broker { |
|
|
|
anonymous = Vec::from_iter(broker.anonymous_connections.keys().cloned()); |
|
|
|
anonymous = Vec::from_iter(broker.anonymous_connections.keys().cloned()); |
|
|
|
} |
|
|
|
} |
|
|
|
for peer_id in peer_ids { |
|
|
|
for peer_id in peer_ids { |
|
|
|
BROKER.write().await.close_peer_connection(&peer_id).await; |
|
|
|
BROKER.write().await.close_peer_connection_x(&peer_id).await; |
|
|
|
} |
|
|
|
} |
|
|
|
for anon in anonymous { |
|
|
|
for anon in anonymous { |
|
|
|
BROKER.write().await.close_anonymous(anon.1, anon.0).await; |
|
|
|
BROKER.write().await.close_anonymous(anon.1, anon.0).await; |
|
|
|
@ -475,7 +478,7 @@ impl Broker { |
|
|
|
lastPeerAdvert: None, |
|
|
|
lastPeerAdvert: None, |
|
|
|
connected, |
|
|
|
connected, |
|
|
|
}; |
|
|
|
}; |
|
|
|
self.peers.insert(remote_peer_id, bpi); |
|
|
|
self.peers.insert(remote_peer_id.to_dh_slice(), bpi); |
|
|
|
|
|
|
|
|
|
|
|
Ok(()) |
|
|
|
Ok(()) |
|
|
|
} |
|
|
|
} |
|
|
|
@ -495,9 +498,6 @@ impl Broker { |
|
|
|
pub async fn connect( |
|
|
|
pub async fn connect( |
|
|
|
&mut self, |
|
|
|
&mut self, |
|
|
|
cnx: Box<dyn IConnect>, |
|
|
|
cnx: Box<dyn IConnect>, |
|
|
|
ip: IP, |
|
|
|
|
|
|
|
port: u16, |
|
|
|
|
|
|
|
core: Option<String>, // the interface used as egress for this connection
|
|
|
|
|
|
|
|
peer_privk: Sensitive<[u8; 32]>, |
|
|
|
peer_privk: Sensitive<[u8; 32]>, |
|
|
|
peer_pubk: PubKey, |
|
|
|
peer_pubk: PubKey, |
|
|
|
remote_peer_id: DirectPeerId, |
|
|
|
remote_peer_id: DirectPeerId, |
|
|
|
@ -513,44 +513,46 @@ impl Broker { |
|
|
|
log_info!("CONNECTING"); |
|
|
|
log_info!("CONNECTING"); |
|
|
|
let mut connection = cnx |
|
|
|
let mut connection = cnx |
|
|
|
.open( |
|
|
|
.open( |
|
|
|
ip, |
|
|
|
config.get_url(), |
|
|
|
port, |
|
|
|
|
|
|
|
Sensitive::<[u8; 32]>::from_slice(peer_privk.deref()), |
|
|
|
Sensitive::<[u8; 32]>::from_slice(peer_privk.deref()), |
|
|
|
peer_pubk, |
|
|
|
peer_pubk, |
|
|
|
remote_peer_id, |
|
|
|
remote_peer_id, |
|
|
|
config, |
|
|
|
config.clone(), |
|
|
|
) |
|
|
|
) |
|
|
|
.await?; |
|
|
|
.await?; |
|
|
|
|
|
|
|
|
|
|
|
let join = connection.take_shutdown(); |
|
|
|
let join = connection.take_shutdown(); |
|
|
|
|
|
|
|
|
|
|
|
let connected = if core.is_some() { |
|
|
|
let connected = match &config { |
|
|
|
|
|
|
|
StartConfig::Core(config) => { |
|
|
|
|
|
|
|
let ip = config.addr.ip.clone(); |
|
|
|
let dc = DirectConnection { |
|
|
|
let dc = DirectConnection { |
|
|
|
ip, |
|
|
|
ip, |
|
|
|
interface: core.clone().unwrap(), |
|
|
|
interface: config.interface.clone(), |
|
|
|
remote_peer_id, |
|
|
|
remote_peer_id, |
|
|
|
tp: connection.transport_protocol(), |
|
|
|
tp: connection.transport_protocol(), |
|
|
|
cnx: connection, |
|
|
|
cnx: connection, |
|
|
|
}; |
|
|
|
}; |
|
|
|
self.direct_connections.insert(ip, dc); |
|
|
|
self.direct_connections.insert(ip, dc); |
|
|
|
PeerConnection::Core(ip) |
|
|
|
PeerConnection::Core(ip) |
|
|
|
} else { |
|
|
|
} |
|
|
|
PeerConnection::Client(connection) |
|
|
|
StartConfig::Client(config) => PeerConnection::Client(connection), |
|
|
|
|
|
|
|
_ => unimplemented!(), |
|
|
|
}; |
|
|
|
}; |
|
|
|
|
|
|
|
|
|
|
|
let bpi = BrokerPeerInfo { |
|
|
|
let bpi = BrokerPeerInfo { |
|
|
|
lastPeerAdvert: None, |
|
|
|
lastPeerAdvert: None, |
|
|
|
connected, |
|
|
|
connected, |
|
|
|
}; |
|
|
|
}; |
|
|
|
self.peers.insert(remote_peer_id, bpi); |
|
|
|
self.peers.insert(remote_peer_id.to_dh_slice(), bpi); |
|
|
|
|
|
|
|
|
|
|
|
async fn watch_close( |
|
|
|
async fn watch_close( |
|
|
|
mut join: Receiver<Either<NetError, PubKey>>, |
|
|
|
mut join: Receiver<Either<NetError, PubKey>>, |
|
|
|
cnx: Box<dyn IConnect>, |
|
|
|
cnx: Box<dyn IConnect>, |
|
|
|
ip: IP, |
|
|
|
|
|
|
|
core: Option<String>, // the interface used as egress for this connection
|
|
|
|
|
|
|
|
peer_privk: Sensitive<[u8; 32]>, |
|
|
|
peer_privk: Sensitive<[u8; 32]>, |
|
|
|
peer_pubkey: PubKey, |
|
|
|
peer_pubkey: PubKey, |
|
|
|
remote_peer_id: DirectPeerId, |
|
|
|
remote_peer_id: DirectPeerId, |
|
|
|
|
|
|
|
config: StartConfig, |
|
|
|
) -> ResultSend<()> { |
|
|
|
) -> ResultSend<()> { |
|
|
|
async move { |
|
|
|
async move { |
|
|
|
let res = join.next().await; |
|
|
|
let res = join.next().await; |
|
|
|
@ -579,16 +581,15 @@ impl Broker { |
|
|
|
spawn_and_log_error(watch_close( |
|
|
|
spawn_and_log_error(watch_close( |
|
|
|
join, |
|
|
|
join, |
|
|
|
cnx, |
|
|
|
cnx, |
|
|
|
ip, |
|
|
|
|
|
|
|
core, |
|
|
|
|
|
|
|
peer_privk, |
|
|
|
peer_privk, |
|
|
|
peer_pubk, |
|
|
|
peer_pubk, |
|
|
|
remote_peer_id, |
|
|
|
remote_peer_id, |
|
|
|
|
|
|
|
config, |
|
|
|
)); |
|
|
|
)); |
|
|
|
Ok(()) |
|
|
|
Ok(()) |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
pub async fn close_peer_connection(&mut self, peer_id: &DirectPeerId) { |
|
|
|
pub async fn close_peer_connection_x(&mut self, peer_id: &X25519PubKey) { |
|
|
|
if let Some(peer) = self.peers.get_mut(peer_id) { |
|
|
|
if let Some(peer) = self.peers.get_mut(peer_id) { |
|
|
|
match &mut peer.connected { |
|
|
|
match &mut peer.connected { |
|
|
|
PeerConnection::Core(_) => { |
|
|
|
PeerConnection::Core(_) => { |
|
|
|
@ -604,6 +605,10 @@ impl Broker { |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
pub async fn close_peer_connection(&mut self, peer_id: &DirectPeerId) { |
|
|
|
|
|
|
|
self.close_peer_connection_x(&peer_id.to_dh_slice()).await |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
pub async fn close_anonymous( |
|
|
|
pub async fn close_anonymous( |
|
|
|
&mut self, |
|
|
|
&mut self, |
|
|
|
remote_bind_address: BindAddress, |
|
|
|
remote_bind_address: BindAddress, |
|
|
|
|
