liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

test fixes

Browse Source
This commit is contained in:
John Smith
2022-06-08 20:07:26 -04:00
parent 69d68e900e
commit 86567ea78d
20 changed files with 135 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
veilid-core/src/network_manager/native/mod.rs
View File

@@ -4,8 +4,8 @@ mod network_udp;
mod protocol;
mod start_protocols;
use super::*;
use crate::intf::*;
use crate::network_manager::*;
use crate::routing_table::*;
use connection_manager::*;
use network_tcp::*;

102
veilid-core/src/network_manager/network_connection.rs
View File

@@ -224,72 +224,68 @@ impl NetworkConnection {
})
};
let timer = MutableFuture::new(new_timer());
unord.push(timer.clone().boxed());
unord.push(system_boxed(timer.clone()));
loop {
// Add another message sender future if necessary
if need_sender {
need_sender = false;
unord.push(
receiver
.recv_async()
.then(|res| async {
match res {
Ok(message) => {
// send the packet
if let Err(e) = Self::send_internal(
&protocol_connection,
stats.clone(),
message,
)
.await
{
// Sending the packet along can fail, if so, this connection is dead
log_net!(debug e);
RecvLoopAction::Finish
} else {
RecvLoopAction::Send
}
}
Err(e) => {
// All senders gone, shouldn't happen since we store one alongside the join handle
log_net!(warn e);
RecvLoopAction::Finish
}
let sender_fut = receiver.recv_async().then(|res| async {
match res {
Ok(message) => {
// send the packet
if let Err(e) = Self::send_internal(
&protocol_connection,
stats.clone(),
message,
)
.await
{
// Sending the packet along can fail, if so, this connection is dead
log_net!(debug e);
RecvLoopAction::Finish
} else {
RecvLoopAction::Send
}
})
.boxed(),
);
}
Err(e) => {
// All senders gone, shouldn't happen since we store one alongside the join handle
log_net!(warn e);
RecvLoopAction::Finish
}
}
});
unord.push(system_boxed(sender_fut));
}
// Add another message receiver future if necessary
if need_receiver {
need_sender = false;
unord.push(
Self::recv_internal(&protocol_connection, stats.clone())
.then(|res| async {
match res {
Ok(message) => {
// Pass received messages up to the network manager for processing
if let Err(e) = network_manager
.on_recv_envelope(message.as_slice(), descriptor)
.await
{
log_net!(error e);
RecvLoopAction::Finish
} else {
RecvLoopAction::Recv
}
}
Err(e) => {
// Connection unable to receive, closed
log_net!(warn e);
need_receiver = false;
let receiver_fut = Self::recv_internal(&protocol_connection, stats.clone())
.then(|res| async {
match res {
Ok(message) => {
// Pass received messages up to the network manager for processing
if let Err(e) = network_manager
.on_recv_envelope(message.as_slice(), descriptor)
.await
{
log_net!(error e);
RecvLoopAction::Finish
} else {
RecvLoopAction::Recv
}
}
})
.boxed(),
);
Err(e) => {
// Connection unable to receive, closed
log_net!(warn e);
RecvLoopAction::Finish
}
}
});
unord.push(system_boxed(receiver_fut));
}
// Process futures

12
veilid-core/src/network_manager/wasm/mod.rs
View File

@@ -1,10 +1,8 @@
mod protocol;
use crate::connection_manager::*;
use crate::network_manager::*;
use super::*;
use crate::routing_table::*;
use crate::intf::*;
use crate::*;
use connection_manager::*;
use protocol::ws::WebsocketProtocolHandler;
pub use protocol::*;
@@ -102,11 +100,11 @@ impl Network {
// Try to send to the exact existing connection if one exists
if let Some(conn) = self.connection_manager().get_connection(descriptor).await {
// connection exists, send over it
conn.send(data).await.map_err(logthru_net!())?;
conn.send_async(data).await.map_err(logthru_net!())?;
// Network accounting
self.network_manager()
.stats_packet_sent(descriptor.remote.to_socket_addr().ip(), data_len as u64);
.stats_packet_sent(descriptor.remote().to_socket_addr().ip(), data_len as u64);
// Data was consumed
Ok(None)
@@ -136,7 +134,7 @@ impl Network {
.get_or_create_connection(None, dial_info.clone())
.await?;
let res = conn.send(data).await.map_err(logthru_net!(error));
let res = conn.send_async(data).await.map_err(logthru_net!(error));
if res.is_ok() {
// Network accounting
self.network_manager()

13
veilid-core/src/network_manager/wasm/protocol/mod.rs
View File

@@ -1,9 +1,8 @@
pub mod wrtc;
pub mod ws;
use crate::network_connection::*;
use super::*;
use crate::xx::*;
use crate::*;
#[derive(Debug)]
pub enum ProtocolNetworkConnection {
@@ -16,7 +15,7 @@ impl ProtocolNetworkConnection {
pub async fn connect(
local_address: Option<SocketAddr>,
dial_info: DialInfo,
) -> Result<NetworkConnection, String> {
) -> Result<ProtocolNetworkConnection, String> {
match dial_info.protocol_type() {
ProtocolType::UDP => {
panic!("UDP dial info is not support on WASM targets");
@@ -46,6 +45,14 @@ impl ProtocolNetworkConnection {
}
}
}
pub fn descriptor(&self) -> ConnectionDescriptor {
match self {
Self::Dummy(d) => d.descriptor(),
Self::Ws(w) => w.descriptor(),
}
}
pub async fn close(&self) -> Result<(), String> {
match self {
Self::Dummy(d) => d.close(),

13
veilid-core/src/network_manager/wasm/protocol/ws.rs
View File

@@ -1,8 +1,4 @@
use crate::intf::*;
use crate::network_connection::*;
use crate::network_manager::MAX_MESSAGE_SIZE;
use crate::*;
use alloc::fmt;
use super::*;
use ws_stream_wasm::*;
use futures_util::{StreamExt, SinkExt};
@@ -104,10 +100,9 @@ impl WebsocketProtocolHandler {
// Make our connection descriptor
Ok(ProtocolNetworkConnection::Ws(WebsocketNetworkConnection::new(ConnectionDescriptor {
local: None,
remote: dial_info.to_peer_address(),
}, wsmeta, wsio)))
Ok(ProtocolNetworkConnection::Ws(WebsocketNetworkConnection::new(ConnectionDescriptor::new_no_local(
dial_info.to_peer_address(),
), wsmeta, wsio)))
}
pub async fn send_unbound_message(dial_info: DialInfo, data: Vec<u8>) -> Result<(), String> {