Jędrzej Stuczyński
0ee387d983
* squashing work on using cancellation in nym crates making nym-task wasm compilable removed sending of status messages replaced TaskManager with ShutdownManager in the validator rewarder additional helpers for ShutdownManager simplified ShutdownToken by removing the name field TaskClient => ShutdownToken within all client tasks wip: remove TaskHandle * track all long-living client tasks * add task tracking for most top level tasks within nym-node * improved default builder * split up cancellation module * module documentation and unit tests * nym node fixes and naming consistency * wasm fixes * assert_eq => assert * wasm fixes and made 'run_until_shutdown' take reference instead of ownership * linux-specific fixes to IpPacketRouter * post rebasing fixes for signing monitor * add ShutdownManager constructor to build it from an external token * applying PR review suggestions
87 lines
3.2 KiB
Rust
87 lines
3.2 KiB
Rust
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
|
|
// JS: I personally don't like this name very much, but could not think of anything better.
|
|
// I will gladly take any suggestions on how to rename this.
|
|
|
|
use crate::error::GatewayClientError;
|
|
use crate::GatewayPacketRouter;
|
|
use futures::channel::mpsc;
|
|
use nym_task::ShutdownToken;
|
|
|
|
pub type MixnetMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
|
|
pub type MixnetMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
|
|
|
|
pub type AcknowledgementSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
|
|
pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
|
|
|
|
#[derive(Clone, Debug)]
|
|
pub struct PacketRouter {
|
|
ack_sender: AcknowledgementSender,
|
|
mixnet_message_sender: MixnetMessageSender,
|
|
shutdown: ShutdownToken,
|
|
}
|
|
|
|
impl PacketRouter {
|
|
pub fn new(
|
|
ack_sender: AcknowledgementSender,
|
|
mixnet_message_sender: MixnetMessageSender,
|
|
shutdown: ShutdownToken,
|
|
) -> Self {
|
|
PacketRouter {
|
|
ack_sender,
|
|
mixnet_message_sender,
|
|
shutdown,
|
|
}
|
|
}
|
|
|
|
pub fn route_mixnet_messages(
|
|
&self,
|
|
received_messages: Vec<Vec<u8>>,
|
|
) -> Result<(), GatewayClientError> {
|
|
if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) {
|
|
// check if the failure is due to the shutdown being in progress and thus the receiver channel
|
|
// having already been dropped
|
|
if self.shutdown.is_cancelled() {
|
|
// This should ideally not happen, but it's ok
|
|
tracing::warn!("Failed to send mixnet messages due to receiver task shutdown");
|
|
return Err(GatewayClientError::ShutdownInProgress);
|
|
}
|
|
// This should never happen during ordinary operation the way it's currently used.
|
|
// Abort to be on the safe side
|
|
panic!("Failed to send mixnet message: {err}");
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub fn route_acks(&self, received_acks: Vec<Vec<u8>>) -> Result<(), GatewayClientError> {
|
|
if let Err(err) = self.ack_sender.unbounded_send(received_acks) {
|
|
// check if the failure is due to the shutdown being in progress and thus the receiver channel
|
|
// having already been dropped
|
|
if self.shutdown.is_cancelled() {
|
|
// This should ideally not happen, but it's ok
|
|
tracing::warn!("Failed to send acks due to receiver task shutdown");
|
|
return Err(GatewayClientError::ShutdownInProgress);
|
|
}
|
|
// This should never happen during ordinary operation the way it's currently used.
|
|
// Abort to be on the safe side
|
|
panic!("Failed to send acks: {err}");
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
impl GatewayPacketRouter for PacketRouter {
|
|
type Error = GatewayClientError;
|
|
|
|
// note: this trait tries to decide whether a given message is an ack or a data message
|
|
|
|
fn route_mixnet_messages(&self, received_messages: Vec<Vec<u8>>) -> Result<(), Self::Error> {
|
|
self.route_mixnet_messages(received_messages)
|
|
}
|
|
|
|
fn route_acks(&self, received_acks: Vec<Vec<u8>>) -> Result<(), Self::Error> {
|
|
self.route_acks(received_acks)
|
|
}
|
|
}
|