mirror of
https://gitlab.com/veilid/veilid.git
synced 2024-12-28 08:49:29 -05:00
lints
This commit is contained in:
parent
d1f728954c
commit
028e02f942
@ -161,22 +161,11 @@ impl AttachmentManager {
|
||||
|
||||
pub fn is_attached(&self) -> bool {
|
||||
let s = self.inner.lock().attachment_machine.state();
|
||||
match s {
|
||||
AttachmentState::Attaching => true,
|
||||
AttachmentState::AttachedWeak => true,
|
||||
AttachmentState::AttachedGood => true,
|
||||
AttachmentState::AttachedStrong => true,
|
||||
AttachmentState::FullyAttached => true,
|
||||
AttachmentState::OverAttached => true,
|
||||
_ => false,
|
||||
}
|
||||
!matches!(s, AttachmentState::Detached | AttachmentState::Detaching)
|
||||
}
|
||||
pub fn is_detached(&self) -> bool {
|
||||
let s = self.inner.lock().attachment_machine.state();
|
||||
match s {
|
||||
AttachmentState::Detached => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(s, AttachmentState::Detached)
|
||||
}
|
||||
|
||||
pub fn get_attach_timestamp(&self) -> Option<u64> {
|
||||
|
@ -89,7 +89,7 @@ where
|
||||
pub async fn consume(&self, input: &T::Input) -> Result<Option<T::Output>, ()> {
|
||||
let current_state = self.inner.lock().state;
|
||||
|
||||
if let Some(new_state) = T::transition(¤t_state, &input) {
|
||||
if let Some(new_state) = T::transition(¤t_state, input) {
|
||||
let output = T::output(¤t_state, input);
|
||||
let old_state = current_state;
|
||||
let (callback, eventual) = {
|
||||
|
@ -25,7 +25,7 @@ impl PartialEq for ConnectionTableEntry {
|
||||
if self.last_message_recv_time != other.last_message_recv_time {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
@ -46,6 +46,11 @@ impl core::fmt::Debug for ConnectionTable {
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ConnectionTable {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
impl ConnectionTable {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
@ -74,7 +79,7 @@ impl ConnectionTable {
|
||||
let timestamp = get_timestamp();
|
||||
|
||||
let entry = ConnectionTableEntry {
|
||||
conn: conn,
|
||||
conn,
|
||||
established_time: timestamp,
|
||||
last_message_sent_time: None,
|
||||
last_message_recv_time: None,
|
||||
@ -90,10 +95,7 @@ impl ConnectionTable {
|
||||
descriptor: &ConnectionDescriptor,
|
||||
) -> Option<ConnectionTableEntry> {
|
||||
let inner = self.inner.lock();
|
||||
match inner.conn_by_addr.get(&descriptor) {
|
||||
Some(v) => Some(v.clone()),
|
||||
None => None,
|
||||
}
|
||||
inner.conn_by_addr.get(descriptor).cloned()
|
||||
}
|
||||
|
||||
pub fn connection_count(&self) -> usize {
|
||||
@ -107,7 +109,7 @@ impl ConnectionTable {
|
||||
) -> Result<ConnectionTableEntry, ()> {
|
||||
let mut inner = self.inner.lock();
|
||||
|
||||
let res = inner.conn_by_addr.remove(&descriptor);
|
||||
let res = inner.conn_by_addr.remove(descriptor);
|
||||
match res {
|
||||
Some(v) => Ok(v.clone()),
|
||||
None => Err(()),
|
||||
|
@ -871,7 +871,7 @@ impl Network {
|
||||
trace!("WS: starting listener at {:?}", listen_address);
|
||||
let (fqdn, port) = split_port(&listen_address)
|
||||
.map_err(|_| "invalid WS listen address, port not specified correctly".to_owned())?;
|
||||
|
||||
let port = port.ok_or_else(|| "port must be specified for WS address".to_owned())?;
|
||||
let _ = self
|
||||
.start_tcp_listener(
|
||||
listen_address.clone(),
|
||||
@ -892,6 +892,8 @@ impl Network {
|
||||
let (public_fqdn, public_port) = split_port(public_address).map_err(|_| {
|
||||
"invalid WS public address, port not specified correctly".to_owned()
|
||||
})?;
|
||||
let public_port = public_port
|
||||
.ok_or_else(|| "port must be specified for public WS address".to_owned())?;
|
||||
|
||||
routing_table.register_public_dial_info(
|
||||
DialInfo::ws(fqdn, public_port, public_fqdn),
|
||||
@ -923,6 +925,7 @@ impl Network {
|
||||
trace!("WSS: starting listener at {}", listen_address);
|
||||
let (fqdn, port) = split_port(&listen_address)
|
||||
.map_err(|_| "invalid WSS listen address, port not specified correctly".to_owned())?;
|
||||
let port = port.ok_or_else(|| "port must be specified for WSS address".to_owned())?;
|
||||
|
||||
let _ = self
|
||||
.start_tcp_listener(
|
||||
@ -944,6 +947,8 @@ impl Network {
|
||||
let (public_fqdn, public_port) = split_port(public_address).map_err(|_| {
|
||||
"invalid WSS public address, port not specified correctly".to_owned()
|
||||
})?;
|
||||
let public_port = public_port
|
||||
.ok_or_else(|| "port must be specified for public WSS address".to_owned())?;
|
||||
|
||||
routing_table.register_public_dial_info(
|
||||
DialInfo::wss(fqdn, public_port, public_fqdn),
|
||||
|
@ -35,7 +35,7 @@ pub struct LeaseManager {
|
||||
impl LeaseManager {
|
||||
fn new_inner(network_manager: NetworkManager) -> LeaseManagerInner {
|
||||
LeaseManagerInner {
|
||||
network_manager: network_manager,
|
||||
network_manager,
|
||||
max_server_signal_leases: 1,
|
||||
max_server_relay_leases: 1,
|
||||
max_client_signal_leases: 1,
|
||||
@ -111,7 +111,7 @@ impl LeaseManager {
|
||||
return false;
|
||||
}
|
||||
let network_class = inner.network_manager.get_network_class();
|
||||
let out = match network_class {
|
||||
match network_class {
|
||||
NetworkClass::Server => true,
|
||||
NetworkClass::Mapped => true,
|
||||
NetworkClass::FullNAT => true,
|
||||
@ -121,8 +121,7 @@ impl LeaseManager {
|
||||
NetworkClass::WebApp => false,
|
||||
NetworkClass::TorWebApp => false,
|
||||
NetworkClass::Invalid => false,
|
||||
};
|
||||
return out;
|
||||
}
|
||||
}
|
||||
pub fn server_will_provide_signal_lease(&self) -> bool {
|
||||
if !self.server_can_provide_signal_lease() {
|
||||
@ -140,7 +139,7 @@ impl LeaseManager {
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
true
|
||||
}
|
||||
|
||||
// Relay leases
|
||||
@ -154,7 +153,7 @@ impl LeaseManager {
|
||||
return false;
|
||||
}
|
||||
let network_class = inner.network_manager.get_network_class();
|
||||
let out = match network_class {
|
||||
match network_class {
|
||||
NetworkClass::Server => true,
|
||||
NetworkClass::Mapped => true,
|
||||
NetworkClass::FullNAT => true,
|
||||
@ -164,9 +163,8 @@ impl LeaseManager {
|
||||
NetworkClass::WebApp => false,
|
||||
NetworkClass::TorWebApp => false,
|
||||
NetworkClass::Invalid => false,
|
||||
};
|
||||
}
|
||||
// xxx: also depends on network strength / bandwidth availability?
|
||||
return out;
|
||||
}
|
||||
pub fn server_will_provide_relay_lease(&self) -> bool {
|
||||
if !self.server_can_provide_relay_lease() {
|
||||
@ -183,6 +181,6 @@ impl LeaseManager {
|
||||
if !routing_table.has_public_dial_info() {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
true
|
||||
}
|
||||
}
|
||||
|
@ -414,7 +414,7 @@ impl NetworkManager {
|
||||
|
||||
// Record the receipt for later
|
||||
let exp_ts = intf::get_timestamp() + expiration_us;
|
||||
let eventual = SingleShotEventual::new(ReceiptEvent::CANCELLED);
|
||||
let eventual = SingleShotEventual::new(ReceiptEvent::Cancelled);
|
||||
let instance = eventual.instance();
|
||||
receipt_manager.record_single_shot_receipt(receipt, exp_ts, eventual);
|
||||
|
||||
@ -462,6 +462,7 @@ impl NetworkManager {
|
||||
// and if so, get the max version we can use
|
||||
let version = if let Some((node_min, node_max)) = node_ref.operate(|e| e.min_max_version())
|
||||
{
|
||||
#[allow(clippy::absurd_extreme_comparisons)]
|
||||
if node_min > MAX_VERSION || node_max < MIN_VERSION {
|
||||
return Err(format!(
|
||||
"can't talk to this node {} because version is unsupported: ({},{})",
|
||||
|
@ -7,9 +7,9 @@ use xx::*;
|
||||
|
||||
#[derive(Clone, Debug, Copy, PartialEq, Eq)]
|
||||
pub enum ReceiptEvent {
|
||||
RETURNED,
|
||||
EXPIRED,
|
||||
CANCELLED,
|
||||
Returned,
|
||||
Expired,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
cfg_if! {
|
||||
@ -114,9 +114,9 @@ impl ReceiptRecord {
|
||||
receipt_callback: impl ReceiptCallback,
|
||||
) -> Self {
|
||||
Self {
|
||||
expiration_ts: expiration_ts,
|
||||
expiration_ts,
|
||||
nonce: receipt.get_nonce(),
|
||||
expected_returns: expected_returns,
|
||||
expected_returns,
|
||||
returns_so_far: 0u32,
|
||||
receipt_callback: ReceiptRecordCallbackType::Normal(Box::new(receipt_callback)),
|
||||
}
|
||||
@ -128,7 +128,7 @@ impl ReceiptRecord {
|
||||
eventual: ReceiptSingleShotType,
|
||||
) -> Self {
|
||||
Self {
|
||||
expiration_ts: expiration_ts,
|
||||
expiration_ts,
|
||||
nonce: receipt.get_nonce(),
|
||||
returns_so_far: 0u32,
|
||||
expected_returns: 1u32,
|
||||
@ -179,7 +179,7 @@ pub struct ReceiptManager {
|
||||
impl ReceiptManager {
|
||||
fn new_inner(network_manager: NetworkManager) -> ReceiptManagerInner {
|
||||
ReceiptManagerInner {
|
||||
network_manager: network_manager,
|
||||
network_manager,
|
||||
receipts_by_nonce: BTreeMap::new(),
|
||||
next_oldest_ts: None,
|
||||
timeout_task: SingleFuture::new(),
|
||||
@ -246,7 +246,7 @@ impl ReceiptManager {
|
||||
let receipt_inner = v.lock();
|
||||
if receipt_inner.expiration_ts <= now {
|
||||
// Expire this receipt
|
||||
expired_nonces.push(k.clone());
|
||||
expired_nonces.push(*k);
|
||||
} else if new_next_oldest_ts.is_none()
|
||||
|| receipt_inner.expiration_ts < new_next_oldest_ts.unwrap()
|
||||
{
|
||||
@ -254,7 +254,7 @@ impl ReceiptManager {
|
||||
new_next_oldest_ts = Some(receipt_inner.expiration_ts);
|
||||
}
|
||||
}
|
||||
if expired_nonces.len() == 0 {
|
||||
if expired_nonces.is_empty() {
|
||||
return;
|
||||
}
|
||||
// Now remove the expired receipts
|
||||
@ -272,7 +272,7 @@ impl ReceiptManager {
|
||||
for expired_record in expired_records {
|
||||
let mut expired_record_mut = expired_record.lock();
|
||||
if let Some(callback) =
|
||||
Self::perform_callback(ReceiptEvent::EXPIRED, &mut expired_record_mut)
|
||||
Self::perform_callback(ReceiptEvent::Expired, &mut expired_record_mut)
|
||||
{
|
||||
callbacks.push(callback)
|
||||
}
|
||||
@ -342,7 +342,7 @@ impl ReceiptManager {
|
||||
fn update_next_oldest_timestamp(inner: &mut ReceiptManagerInner) {
|
||||
// Update the next oldest timestamp
|
||||
let mut new_next_oldest_ts: Option<u64> = None;
|
||||
for (_, v) in &inner.receipts_by_nonce {
|
||||
for v in inner.receipts_by_nonce.values() {
|
||||
let receipt_inner = v.lock();
|
||||
if new_next_oldest_ts.is_none()
|
||||
|| receipt_inner.expiration_ts < new_next_oldest_ts.unwrap()
|
||||
@ -372,7 +372,7 @@ impl ReceiptManager {
|
||||
// Generate a cancelled callback
|
||||
let callback_future = {
|
||||
let mut record_mut = record.lock();
|
||||
Self::perform_callback(ReceiptEvent::CANCELLED, &mut record_mut)
|
||||
Self::perform_callback(ReceiptEvent::Cancelled, &mut record_mut)
|
||||
};
|
||||
|
||||
// Issue the callback
|
||||
@ -397,7 +397,7 @@ impl ReceiptManager {
|
||||
// Generate the callback future
|
||||
let mut record_mut = record.lock();
|
||||
record_mut.returns_so_far += 1;
|
||||
let callback_future = Self::perform_callback(ReceiptEvent::RETURNED, &mut record_mut);
|
||||
let callback_future = Self::perform_callback(ReceiptEvent::Returned, &mut record_mut);
|
||||
|
||||
// Remove the record if we're done
|
||||
if record_mut.returns_so_far == record_mut.expected_returns {
|
||||
|
@ -24,6 +24,6 @@ pub fn decode_node_dial_info_single(
|
||||
|
||||
Ok(NodeDialInfoSingle {
|
||||
node_id: NodeId::new(node_id),
|
||||
dial_info: dial_info,
|
||||
dial_info,
|
||||
})
|
||||
}
|
||||
|
@ -45,6 +45,6 @@ pub fn decode_peer_info(reader: &veilid_capnp::peer_info::Reader) -> Result<Peer
|
||||
}
|
||||
Ok(PeerInfo {
|
||||
node_id: NodeId::new(decode_public_key(&nid_reader)),
|
||||
dial_infos: dial_infos,
|
||||
dial_infos,
|
||||
})
|
||||
}
|
||||
|
@ -67,7 +67,7 @@ pub fn encode_route_hop(
|
||||
)?;
|
||||
if let Some(rhd) = &route_hop.next_hop {
|
||||
let mut rhd_builder = builder.reborrow().init_next_hop();
|
||||
encode_route_hop_data(&rhd, &mut rhd_builder)?;
|
||||
encode_route_hop_data(rhd, &mut rhd_builder)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@ -83,7 +83,7 @@ pub fn encode_private_route(
|
||||
builder.set_hop_count(private_route.hop_count);
|
||||
if let Some(rh) = &private_route.hops {
|
||||
let mut rh_builder = builder.reborrow().init_first_hop();
|
||||
encode_route_hop(&rh, &mut rh_builder)?;
|
||||
encode_route_hop(rh, &mut rh_builder)?;
|
||||
};
|
||||
|
||||
Ok(())
|
||||
@ -102,11 +102,11 @@ pub fn encode_safety_route(
|
||||
match &safety_route.hops {
|
||||
SafetyRouteHops::Data(rhd) => {
|
||||
let mut rhd_builder = h_builder.init_data();
|
||||
encode_route_hop_data(&rhd, &mut rhd_builder)?;
|
||||
encode_route_hop_data(rhd, &mut rhd_builder)?;
|
||||
}
|
||||
SafetyRouteHops::Private(pr) => {
|
||||
let mut pr_builder = h_builder.init_private();
|
||||
encode_private_route(&pr, &mut pr_builder)?;
|
||||
encode_private_route(pr, &mut pr_builder)?;
|
||||
}
|
||||
};
|
||||
|
||||
@ -129,10 +129,7 @@ pub fn decode_route_hop_data(
|
||||
.map_err(map_error_internal!("invalid blob in route hop data"))?
|
||||
.to_vec();
|
||||
|
||||
Ok(RouteHopData {
|
||||
nonce: nonce,
|
||||
blob: blob,
|
||||
})
|
||||
Ok(RouteHopData { nonce, blob })
|
||||
}
|
||||
|
||||
pub fn decode_route_hop(reader: &veilid_capnp::route_hop::Reader) -> Result<RouteHop, RPCError> {
|
||||
@ -153,8 +150,8 @@ pub fn decode_route_hop(reader: &veilid_capnp::route_hop::Reader) -> Result<Rout
|
||||
};
|
||||
|
||||
Ok(RouteHop {
|
||||
dial_info: dial_info,
|
||||
next_hop: next_hop,
|
||||
dial_info,
|
||||
next_hop,
|
||||
})
|
||||
}
|
||||
|
||||
@ -177,9 +174,9 @@ pub fn decode_private_route(
|
||||
};
|
||||
|
||||
Ok(PrivateRoute {
|
||||
public_key: public_key,
|
||||
hop_count: hop_count,
|
||||
hops: hops,
|
||||
public_key,
|
||||
hop_count,
|
||||
hops,
|
||||
})
|
||||
}
|
||||
|
||||
@ -205,8 +202,8 @@ pub fn decode_safety_route(
|
||||
};
|
||||
|
||||
Ok(SafetyRoute {
|
||||
public_key: public_key,
|
||||
hop_count: hop_count,
|
||||
hops: hops,
|
||||
public_key,
|
||||
hop_count,
|
||||
hops,
|
||||
})
|
||||
}
|
||||
|
@ -28,7 +28,5 @@ pub fn decode_sender_info(
|
||||
} else {
|
||||
None
|
||||
};
|
||||
Ok(SenderInfo {
|
||||
socket_address: socket_address,
|
||||
})
|
||||
Ok(SenderInfo { socket_address })
|
||||
}
|
||||
|
@ -116,7 +116,7 @@ impl RespondTo {
|
||||
}
|
||||
Self::PrivateRoute(pr) => {
|
||||
let mut pr_builder = builder.reborrow().init_private_route();
|
||||
encode_private_route(&pr, &mut pr_builder)?;
|
||||
encode_private_route(pr, &mut pr_builder)?;
|
||||
}
|
||||
};
|
||||
Ok(())
|
||||
@ -137,7 +137,7 @@ struct RPCMessageData {
|
||||
}
|
||||
|
||||
impl ReaderSegments for RPCMessageData {
|
||||
fn get_segment<'a>(&'a self, idx: u32) -> Option<&'a [u8]> {
|
||||
fn get_segment(&self, idx: u32) -> Option<&[u8]> {
|
||||
if idx > 0 {
|
||||
None
|
||||
} else {
|
||||
@ -539,12 +539,13 @@ impl RPCProcessor {
|
||||
) -> Result<(), RPCError> {
|
||||
let eventual = {
|
||||
let mut inner = self.inner.lock();
|
||||
inner.waiting_rpc_table.remove(&op_id)
|
||||
inner
|
||||
.waiting_rpc_table
|
||||
.remove(&op_id)
|
||||
.ok_or_else(|| rpc_error_internal("Unmatched operation id"))?
|
||||
};
|
||||
match eventual {
|
||||
None => Err(rpc_error_internal("Unmatched operation id")),
|
||||
Some(e) => Ok(e.resolve(rpcreader).await),
|
||||
}
|
||||
eventual.resolve(rpcreader).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// wait for reply
|
||||
@ -661,11 +662,11 @@ impl RPCProcessor {
|
||||
out_node_id = sr
|
||||
.hops
|
||||
.first()
|
||||
.ok_or(rpc_error_internal("no hop in safety route"))?
|
||||
.ok_or_else(|| rpc_error_internal("no hop in safety route"))?
|
||||
.dial_info
|
||||
.node_id
|
||||
.key;
|
||||
out = self.wrap_with_route(Some(&sr), private_route, message_vec)?;
|
||||
out = self.wrap_with_route(Some(sr), private_route, message_vec)?;
|
||||
hopcount = 1 + sr.hops.len();
|
||||
}
|
||||
};
|
||||
@ -700,11 +701,11 @@ impl RPCProcessor {
|
||||
let out_node_id = sr
|
||||
.hops
|
||||
.first()
|
||||
.ok_or(rpc_error_internal("no hop in safety route"))?
|
||||
.ok_or_else(|| rpc_error_internal("no hop in safety route"))?
|
||||
.dial_info
|
||||
.node_id
|
||||
.key;
|
||||
out = self.wrap_with_route(Some(&sr), pr_reader, message_vec)?;
|
||||
out = self.wrap_with_route(Some(sr), pr_reader, message_vec)?;
|
||||
out_node_id
|
||||
}
|
||||
}
|
||||
@ -746,7 +747,7 @@ impl RPCProcessor {
|
||||
.network_manager()
|
||||
.send_envelope(node_ref.clone(), out)
|
||||
.await
|
||||
.map_err(|e| RPCError::Internal(e))
|
||||
.map_err(RPCError::Internal)
|
||||
{
|
||||
// Make sure to clean up op id waiter in case of error
|
||||
if eventual.is_some() {
|
||||
@ -851,7 +852,7 @@ impl RPCProcessor {
|
||||
out_node_id = sr
|
||||
.hops
|
||||
.first()
|
||||
.ok_or(rpc_error_internal("no hop in safety route"))?
|
||||
.ok_or_else(|| rpc_error_internal("no hop in safety route"))?
|
||||
.dial_info
|
||||
.node_id
|
||||
.key;
|
||||
@ -869,8 +870,7 @@ impl RPCProcessor {
|
||||
};
|
||||
|
||||
// Reply with 'route' operation
|
||||
out =
|
||||
self.wrap_with_route(safety_route_spec.clone(), private_route, reply_vec)?;
|
||||
out = self.wrap_with_route(safety_route_spec, private_route, reply_vec)?;
|
||||
out_node_id = match safety_route_spec {
|
||||
None => {
|
||||
// If no safety route, the first node is the first hop of the private route
|
||||
@ -891,7 +891,7 @@ impl RPCProcessor {
|
||||
// If safety route is in use, first node is the first hop of the safety route
|
||||
sr.hops
|
||||
.first()
|
||||
.ok_or(rpc_error_internal("no hop in safety route"))?
|
||||
.ok_or_else(|| rpc_error_internal("no hop in safety route"))?
|
||||
.dial_info
|
||||
.node_id
|
||||
.key
|
||||
@ -919,7 +919,7 @@ impl RPCProcessor {
|
||||
self.network_manager()
|
||||
.send_envelope(node_ref.clone(), out)
|
||||
.await
|
||||
.map_err(|e| RPCError::Internal(e))?;
|
||||
.map_err(RPCError::Internal)?;
|
||||
|
||||
// Reply successfully sent
|
||||
let send_ts = get_timestamp();
|
||||
@ -971,7 +971,7 @@ impl RPCProcessor {
|
||||
}
|
||||
// xxx: bandwidth limiting here, don't commit to doing info redirects if our network quality sucks
|
||||
|
||||
return true;
|
||||
true
|
||||
}
|
||||
//////////////////////////////////////////////////////////////////////
|
||||
|
||||
@ -1094,7 +1094,7 @@ impl RPCProcessor {
|
||||
let routing_table = self.routing_table();
|
||||
let protocol_address_type = dial_info.protocol_address_type();
|
||||
let peers = routing_table.get_fast_nodes_of_type(protocol_address_type);
|
||||
if peers.len() == 0 {
|
||||
if peers.is_empty() {
|
||||
return Err(rpc_error_internal(format!(
|
||||
"no peers of type '{:?}'",
|
||||
protocol_address_type
|
||||
@ -1222,9 +1222,9 @@ impl RPCProcessor {
|
||||
.try_into()
|
||||
.map_err(map_error_internal!("invalid closest nodes list length"))?,
|
||||
);
|
||||
for i in 0..closest_nodes.len() {
|
||||
for (i, closest_node) in closest_nodes.iter().enumerate() {
|
||||
let mut pi_builder = peers_builder.reborrow().get(i as u32);
|
||||
encode_peer_info(&closest_nodes[i], &mut pi_builder)?;
|
||||
encode_peer_info(closest_node, &mut pi_builder)?;
|
||||
}
|
||||
reply_msg.into_reader()
|
||||
};
|
||||
@ -1440,19 +1440,9 @@ impl RPCProcessor {
|
||||
}
|
||||
|
||||
async fn rpc_worker(self, receiver: Receiver<RPCMessage>) {
|
||||
loop {
|
||||
let msg = match receiver.recv().await {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
break;
|
||||
}
|
||||
};
|
||||
|
||||
match self.process_rpc_message(msg).await {
|
||||
Ok(_) => (),
|
||||
Err(e) => {
|
||||
error!("Couldn't process rpc message: {}", e);
|
||||
}
|
||||
while let Ok(msg) = receiver.recv().await {
|
||||
if let Err(e) = self.process_rpc_message(msg).await {
|
||||
error!("Couldn't process rpc message: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1651,9 +1641,9 @@ impl RPCProcessor {
|
||||
|
||||
// Wait for receipt
|
||||
match eventual_value.await {
|
||||
ReceiptEvent::RETURNED => Ok(true),
|
||||
ReceiptEvent::EXPIRED => Ok(false),
|
||||
ReceiptEvent::CANCELLED => Err(rpc_error_internal(
|
||||
ReceiptEvent::Returned => Ok(true),
|
||||
ReceiptEvent::Expired => Ok(false),
|
||||
ReceiptEvent::Cancelled => Err(rpc_error_internal(
|
||||
"receipt was dropped before expiration".to_owned(),
|
||||
)),
|
||||
}
|
||||
@ -1685,7 +1675,7 @@ impl RPCProcessor {
|
||||
} else {
|
||||
PeerScope::All
|
||||
});
|
||||
if own_peer_info.dial_infos.len() == 0 {
|
||||
if own_peer_info.dial_infos.is_empty() {
|
||||
return Err(rpc_error_internal("No valid public dial info for own node"));
|
||||
}
|
||||
|
||||
|
@ -136,5 +136,5 @@ pub async fn test_all() {
|
||||
test_enc_dec().await;
|
||||
test_dh(crypto).await;
|
||||
shutdown(api.clone()).await;
|
||||
assert_eq!(api.is_shutdown(), true);
|
||||
assert!(api.is_shutdown());
|
||||
}
|
||||
|
@ -1,3 +1,5 @@
|
||||
#![allow(clippy::bool_assert_comparison)]
|
||||
|
||||
use crate::dht::key;
|
||||
use crate::xx::*;
|
||||
use core::convert::TryFrom;
|
||||
@ -104,9 +106,9 @@ pub async fn test_key_conversions() {
|
||||
// Test default key
|
||||
let (dht_key, dht_key_secret) = (key::DHTKey::default(), key::DHTKeySecret::default());
|
||||
assert_eq!(dht_key.bytes, EMPTY_KEY);
|
||||
assert_eq!(dht_key.valid, false);
|
||||
assert!(!dht_key.valid);
|
||||
assert_eq!(dht_key_secret.bytes, EMPTY_KEY_SECRET);
|
||||
assert_eq!(dht_key_secret.valid, false);
|
||||
assert!(!dht_key_secret.valid);
|
||||
let dht_key_string = String::from(&dht_key);
|
||||
trace!("dht_key_string: {:?}", dht_key_string);
|
||||
let dht_key_string2 = String::from(&dht_key);
|
||||
|
@ -152,8 +152,7 @@ pub async fn test_cbor(ts: TableStore) {
|
||||
let d = match db.load_cbor::<key::DHTKey>(0, b"asdf").await {
|
||||
Ok(x) => x,
|
||||
Err(e) => {
|
||||
assert!(false, "couldn't decode cbor: {}", e);
|
||||
return;
|
||||
panic!("couldn't decode cbor: {}", e);
|
||||
}
|
||||
};
|
||||
assert_eq!(d, Some(dht_key), "keys should be equal");
|
||||
|
@ -1,3 +1,5 @@
|
||||
#![allow(clippy::bool_assert_comparison)]
|
||||
|
||||
use crate::xx::*;
|
||||
use crate::*;
|
||||
cfg_if! {
|
||||
@ -227,7 +229,7 @@ pub async fn test_config() {
|
||||
Ok(()) => (),
|
||||
Err(e) => {
|
||||
error!("Error: {}", e);
|
||||
assert!(false);
|
||||
unreachable!();
|
||||
}
|
||||
}
|
||||
let inner = vc.get();
|
||||
|
@ -119,7 +119,7 @@ cfg_if! {
|
||||
|
||||
static SETUP_ONCE: Once = Once::new();
|
||||
|
||||
pub fn setup() -> () {
|
||||
pub fn setup() {
|
||||
SETUP_ONCE.call_once(|| {
|
||||
let mut cb = ConfigBuilder::new();
|
||||
cb.add_filter_ignore_str("async_std");
|
||||
|
@ -21,7 +21,7 @@ pub struct NodeId {
|
||||
}
|
||||
impl NodeId {
|
||||
pub fn new(key: DHTKey) -> Self {
|
||||
Self { key: key }
|
||||
Self { key }
|
||||
}
|
||||
}
|
||||
|
||||
@ -32,14 +32,11 @@ pub struct ValueKey {
|
||||
}
|
||||
impl ValueKey {
|
||||
pub fn new(key: DHTKey) -> Self {
|
||||
Self {
|
||||
key: key,
|
||||
subkey: None,
|
||||
}
|
||||
Self { key, subkey: None }
|
||||
}
|
||||
pub fn new_subkey(key: DHTKey, subkey: String) -> Self {
|
||||
Self {
|
||||
key: key,
|
||||
key,
|
||||
subkey: if subkey.len() == 0 {
|
||||
None
|
||||
} else {
|
||||
@ -55,7 +52,7 @@ pub struct BlockId {
|
||||
}
|
||||
impl BlockId {
|
||||
pub fn new(key: DHTKey) -> Self {
|
||||
Self { key: key }
|
||||
Self { key }
|
||||
}
|
||||
}
|
||||
|
||||
@ -220,33 +217,19 @@ impl DialInfo {
|
||||
if let Address::Hostname(_) = address {
|
||||
panic!("invalid address type for protocol")
|
||||
}
|
||||
Self::UDP(DialInfoUDP {
|
||||
address: address,
|
||||
port: port,
|
||||
})
|
||||
Self::UDP(DialInfoUDP { address, port })
|
||||
}
|
||||
pub fn tcp(address: Address, port: u16) -> Self {
|
||||
if let Address::Hostname(_) = address {
|
||||
panic!("invalid address type for protocol")
|
||||
}
|
||||
Self::TCP(DialInfoTCP {
|
||||
address: address,
|
||||
port: port,
|
||||
})
|
||||
Self::TCP(DialInfoTCP { address, port })
|
||||
}
|
||||
pub fn ws(fqdn: String, port: u16, path: String) -> Self {
|
||||
Self::WS(DialInfoWS {
|
||||
fqdn: fqdn,
|
||||
port: port,
|
||||
path: path,
|
||||
})
|
||||
Self::WS(DialInfoWS { fqdn, port, path })
|
||||
}
|
||||
pub fn wss(fqdn: String, port: u16, path: String) -> Self {
|
||||
Self::WSS(DialInfoWSS {
|
||||
fqdn: fqdn,
|
||||
port: port,
|
||||
path: path,
|
||||
})
|
||||
Self::WSS(DialInfoWSS { fqdn, port, path })
|
||||
}
|
||||
pub fn protocol_type(&self) -> ProtocolType {
|
||||
match self {
|
||||
@ -518,9 +501,9 @@ pub struct PeerAddress {
|
||||
impl PeerAddress {
|
||||
pub fn new(address: Address, port: u16, protocol_type: ProtocolType) -> Self {
|
||||
Self {
|
||||
address: address,
|
||||
port: port,
|
||||
protocol_type: protocol_type,
|
||||
address,
|
||||
port,
|
||||
protocol_type,
|
||||
}
|
||||
}
|
||||
|
||||
@ -554,13 +537,13 @@ pub struct ConnectionDescriptor {
|
||||
impl ConnectionDescriptor {
|
||||
pub fn new(remote: PeerAddress, local: SocketAddr) -> Self {
|
||||
Self {
|
||||
remote: remote,
|
||||
remote,
|
||||
local: Some(local),
|
||||
}
|
||||
}
|
||||
pub fn new_no_local(remote: PeerAddress) -> Self {
|
||||
Self {
|
||||
remote: remote,
|
||||
remote,
|
||||
local: None,
|
||||
}
|
||||
}
|
||||
@ -642,7 +625,7 @@ impl core::str::FromStr for NodeDialInfoSingle {
|
||||
|
||||
// build NodeDialInfoSingle
|
||||
Ok(NodeDialInfoSingle {
|
||||
node_id: node_id,
|
||||
node_id,
|
||||
dial_info: match proto {
|
||||
ProtocolType::UDP => DialInfo::udp(address, port),
|
||||
ProtocolType::TCP => DialInfo::tcp(address, port),
|
||||
@ -853,10 +836,7 @@ pub struct RoutingContext {
|
||||
impl RoutingContext {
|
||||
fn new(api: VeilidAPI, options: RoutingContextOptions) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(Mutex::new(RoutingContextInner {
|
||||
api: api,
|
||||
options: options,
|
||||
})),
|
||||
inner: Arc::new(Mutex::new(RoutingContextInner { api, options })),
|
||||
}
|
||||
}
|
||||
|
||||
@ -933,7 +913,7 @@ impl VeilidAPI {
|
||||
inner: Arc::new(Mutex::new(VeilidAPIInner {
|
||||
config: attachment_manager.config(),
|
||||
attachment_manager: attachment_manager.clone(),
|
||||
core: core,
|
||||
core,
|
||||
network_manager: attachment_manager.network_manager(),
|
||||
is_shutdown: false,
|
||||
})),
|
||||
@ -948,9 +928,9 @@ impl VeilidAPI {
|
||||
self.inner.lock().attachment_manager.clone()
|
||||
}
|
||||
|
||||
fn network_manager(&self) -> NetworkManager {
|
||||
self.inner.lock().network_manager.clone()
|
||||
}
|
||||
// fn network_manager(&self) -> NetworkManager {
|
||||
// self.inner.lock().network_manager.clone()
|
||||
// }
|
||||
|
||||
fn rpc_processor(&self) -> RPCProcessor {
|
||||
self.inner.lock().network_manager.rpc_processor()
|
||||
|
@ -164,6 +164,11 @@ pub struct VeilidConfig {
|
||||
inner: Arc<RwLock<VeilidConfigInner>>,
|
||||
}
|
||||
|
||||
impl Default for VeilidConfig {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
impl VeilidConfig {
|
||||
fn new_inner() -> VeilidConfigInner {
|
||||
VeilidConfigInner::default()
|
||||
|
@ -43,6 +43,12 @@ pub struct VeilidCore {
|
||||
inner: Arc<Mutex<VeilidCoreInner>>,
|
||||
}
|
||||
|
||||
impl Default for VeilidCore {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl VeilidCore {
|
||||
fn new_inner() -> VeilidCoreInner {
|
||||
VeilidCoreInner {
|
||||
@ -58,18 +64,18 @@ impl VeilidCore {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn config(&self) -> VeilidConfig {
|
||||
self.inner.lock().config.as_ref().unwrap().clone()
|
||||
}
|
||||
// pub(crate) fn config(&self) -> VeilidConfig {
|
||||
// self.inner.lock().config.as_ref().unwrap().clone()
|
||||
// }
|
||||
|
||||
pub(crate) fn attachment_manager(&self) -> AttachmentManager {
|
||||
self.inner
|
||||
.lock()
|
||||
.attachment_manager
|
||||
.as_ref()
|
||||
.unwrap()
|
||||
.clone()
|
||||
}
|
||||
// pub(crate) fn attachment_manager(&self) -> AttachmentManager {
|
||||
// self.inner
|
||||
// .lock()
|
||||
// .attachment_manager
|
||||
// .as_ref()
|
||||
// .unwrap()
|
||||
// .clone()
|
||||
// }
|
||||
|
||||
pub(crate) fn table_store(&self) -> TableStore {
|
||||
self.inner.lock().table_store.as_ref().unwrap().clone()
|
||||
@ -119,8 +125,8 @@ impl VeilidCore {
|
||||
.init(Arc::new(
|
||||
move |old_state: AttachmentState, new_state: AttachmentState| {
|
||||
cb(VeilidStateChange::Attachment {
|
||||
old_state: old_state,
|
||||
new_state: new_state,
|
||||
old_state,
|
||||
new_state,
|
||||
})
|
||||
},
|
||||
))
|
||||
|
@ -23,6 +23,6 @@ impl RngCore for VeilidRng {
|
||||
}
|
||||
|
||||
fn try_fill_bytes(&mut self, dest: &mut [u8]) -> Result<(), Error> {
|
||||
intf::random_bytes(dest).map_err(|err| Error::new(err))
|
||||
intf::random_bytes(dest).map_err(Error::new)
|
||||
}
|
||||
}
|
||||
|
@ -26,6 +26,12 @@ impl EventualBase for Eventual {
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for Eventual {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Eventual {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
@ -39,7 +45,7 @@ impl Eventual {
|
||||
{
|
||||
EventualFutureClone {
|
||||
id: None,
|
||||
value: value,
|
||||
value,
|
||||
eventual: self.clone(),
|
||||
}
|
||||
}
|
||||
|
@ -34,7 +34,7 @@ impl<T> EventualBaseInner<T> {
|
||||
self.wakers.remove(&id);
|
||||
// See if we should complete the EventualResolvedFutures
|
||||
let mut resolved_waker_list = Vec::new();
|
||||
if self.wakers.len() == 0 && self.resolved.is_some() {
|
||||
if self.wakers.is_empty() && self.resolved.is_some() {
|
||||
for w in &self.resolved_wakers {
|
||||
resolved_waker_list.push(w.1.clone());
|
||||
}
|
||||
@ -92,26 +92,25 @@ impl<T> EventualBaseInner<T> {
|
||||
self.resolved_freelist.clear();
|
||||
}
|
||||
|
||||
pub(super) fn try_reset(&mut self) -> Result<(), ()> {
|
||||
if self.wakers.len() != 0 {
|
||||
return Err(());
|
||||
pub(super) fn try_reset(&mut self) -> Result<(), String> {
|
||||
if !self.wakers.is_empty() {
|
||||
return Err("Wakers not empty during reset".to_owned());
|
||||
}
|
||||
if self.resolved_wakers.len() != 0 {
|
||||
return Err(());
|
||||
if !self.resolved_wakers.is_empty() {
|
||||
return Err("Resolved wakers not empty during reset".to_owned());
|
||||
}
|
||||
self.reset();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// Resolved future helpers
|
||||
#[must_use]
|
||||
pub(super) fn resolved_poll(
|
||||
&mut self,
|
||||
id: &mut Option<usize>,
|
||||
cx: &mut task::Context<'_>,
|
||||
) -> task::Poll<()> {
|
||||
// If there are any instance futures still waiting, we resolution isn't finished
|
||||
if self.wakers.len() != 0 {
|
||||
if !self.wakers.is_empty() {
|
||||
if id.is_none() {
|
||||
*id = Some(self.insert_resolved_waker(cx.waker().clone()));
|
||||
}
|
||||
@ -137,12 +136,10 @@ impl<T> EventualBaseInner<T> {
|
||||
*id = Some(self.insert_waker(cx.waker().clone()));
|
||||
}
|
||||
None
|
||||
} else if let Some(id) = id.take() {
|
||||
Some(self.remove_waker(id))
|
||||
} else {
|
||||
if let Some(id) = id.take() {
|
||||
Some(self.remove_waker(id))
|
||||
} else {
|
||||
Some(Vec::new())
|
||||
}
|
||||
Some(Vec::new())
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -202,7 +199,7 @@ pub trait EventualCommon: EventualBase {
|
||||
self.base_inner().reset()
|
||||
}
|
||||
|
||||
fn try_reset(&self) -> Result<(), ()> {
|
||||
fn try_reset(&self) -> Result<(), String> {
|
||||
self.base_inner().try_reset()
|
||||
}
|
||||
}
|
||||
|
@ -26,6 +26,12 @@ impl<T: Unpin> EventualBase for EventualValue<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Unpin> Default for EventualValue<T> {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Unpin> EventualValue<T> {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
|
@ -26,6 +26,12 @@ impl<T: Unpin + Clone> EventualBase for EventualValueClone<T> {
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Unpin + Clone> Default for EventualValueClone<T> {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl<T: Unpin + Clone> EventualValueClone<T> {
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
|
@ -8,10 +8,7 @@ pub struct IpAddrPort {
|
||||
|
||||
impl IpAddrPort {
|
||||
pub fn new(addr: IpAddr, port: u16) -> Self {
|
||||
Self {
|
||||
addr: addr,
|
||||
port: port,
|
||||
}
|
||||
Self { addr, port }
|
||||
}
|
||||
pub fn addr(&self) -> &IpAddr {
|
||||
&self.addr
|
||||
|
@ -63,17 +63,14 @@ pub fn ipv4addr_is_loopback(addr: &Ipv4Addr) -> bool {
|
||||
pub fn ipv4addr_is_private(addr: &Ipv4Addr) -> bool {
|
||||
match addr.octets() {
|
||||
[10, ..] => true,
|
||||
[172, b, ..] if b >= 16 && b <= 31 => true,
|
||||
[172, b, ..] if (16..=31).contains(&b) => true,
|
||||
[192, 168, ..] => true,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn ipv4addr_is_link_local(addr: &Ipv4Addr) -> bool {
|
||||
match addr.octets() {
|
||||
[169, 254, ..] => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(addr.octets(), [169, 254, ..])
|
||||
}
|
||||
|
||||
pub fn ipv4addr_is_global(addr: &Ipv4Addr) -> bool {
|
||||
@ -120,12 +117,10 @@ pub fn ipv4addr_is_broadcast(addr: &Ipv4Addr) -> bool {
|
||||
}
|
||||
|
||||
pub fn ipv4addr_is_documentation(addr: &Ipv4Addr) -> bool {
|
||||
match addr.octets() {
|
||||
[192, 0, 2, _] => true,
|
||||
[198, 51, 100, _] => true,
|
||||
[203, 0, 113, _] => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(
|
||||
addr.octets(),
|
||||
[192, 0, 2, _] | [198, 51, 100, _] | [203, 0, 113, _]
|
||||
)
|
||||
}
|
||||
|
||||
pub fn ipv6addr_is_unspecified(addr: &Ipv6Addr) -> bool {
|
||||
@ -149,10 +144,10 @@ pub fn ipv6addr_is_unique_local(addr: &Ipv6Addr) -> bool {
|
||||
}
|
||||
|
||||
pub fn ipv6addr_is_unicast_link_local_strict(addr: &Ipv6Addr) -> bool {
|
||||
(addr.segments()[0] & 0xffff) == 0xfe80
|
||||
&& (addr.segments()[1] & 0xffff) == 0
|
||||
&& (addr.segments()[2] & 0xffff) == 0
|
||||
&& (addr.segments()[3] & 0xffff) == 0
|
||||
addr.segments()[0] == 0xfe80
|
||||
&& addr.segments()[1] == 0
|
||||
&& addr.segments()[2] == 0
|
||||
&& addr.segments()[3] == 0
|
||||
}
|
||||
|
||||
pub fn ipv6addr_is_unicast_link_local(addr: &Ipv6Addr) -> bool {
|
||||
|
@ -56,8 +56,8 @@ where
|
||||
|
||||
fn unlock(&self, jh: Option<JoinHandle<T>>) {
|
||||
let mut inner = self.inner.lock();
|
||||
assert_eq!(inner.locked, true);
|
||||
assert_eq!(inner.join_handle.is_none(), true);
|
||||
assert!(inner.locked);
|
||||
assert!(inner.join_handle.is_none());
|
||||
inner.locked = false;
|
||||
inner.join_handle = jh;
|
||||
}
|
||||
|
@ -24,7 +24,7 @@ where
|
||||
pub fn new(drop_value: T) -> Self {
|
||||
Self {
|
||||
eventual: EventualValueClone::new(),
|
||||
drop_value: drop_value,
|
||||
drop_value,
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -24,7 +24,7 @@ impl TickTask {
|
||||
pub fn new_us(tick_period_us: u64) -> Self {
|
||||
Self {
|
||||
last_timestamp_us: AtomicU64::new(0),
|
||||
tick_period_us: tick_period_us,
|
||||
tick_period_us,
|
||||
routine: OnceCell::new(),
|
||||
single_future: SingleFuture::new(),
|
||||
}
|
||||
|
@ -1,20 +1,22 @@
|
||||
use crate::xx::*;
|
||||
use alloc::string::ToString;
|
||||
|
||||
pub fn split_port(name: &str) -> Result<(String, u16), ()> {
|
||||
pub fn split_port(name: &str) -> Result<(String, Option<u16>), String> {
|
||||
if let Some(split) = name.rfind(':') {
|
||||
let hoststr = &name[0..split];
|
||||
let portstr = &name[split + 1..];
|
||||
let port: u16 = portstr.parse::<u16>().map_err(drop)?;
|
||||
let port: u16 = portstr
|
||||
.parse::<u16>()
|
||||
.map_err(|e| format!("Invalid port: {}", e))?;
|
||||
|
||||
Ok((hoststr.to_string(), port))
|
||||
Ok((hoststr.to_string(), Some(port)))
|
||||
} else {
|
||||
Err(())
|
||||
Ok((name.to_string(), None))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn prepend_slash(s: String) -> String {
|
||||
if s.starts_with("/") {
|
||||
if s.starts_with('/') {
|
||||
return s;
|
||||
}
|
||||
let mut out = "/".to_owned();
|
||||
|
Loading…
Reference in New Issue
Block a user