Struct Channel

pub struct Channel {
    reader: Mutex<ReadHalf<Box<dyn PtStream>>>,
    writer: Mutex<WriteHalf<Box<dyn PtStream>>>,
    message_subsystem: MessageSubsystem,
    stop_publisher: PublisherPtr<Error>,
    receive_task: StoppableTaskPtr,
    stopped: AtomicBool,
    pub(super) session: SessionWeakPtr,
    pub version: OnceCell<Arc<VersionMessage>>,
    pub info: ChannelInfo,
    metering_map: Mutex<HashMap<String, MeteringQueue>>,
}

Async channel for communication between nodes.

Fields

reader: Mutex<ReadHalf<Box<dyn PtStream>>>

The reading half of the transport stream

writer: Mutex<WriteHalf<Box<dyn PtStream>>>

The writing half of the transport stream

message_subsystem: MessageSubsystem

The message subsystem instance for this channel

stop_publisher: PublisherPtr<Error>

Publisher listening for stop signal for closing this channel

receive_task: StoppableTaskPtr

Task that is listening for the stop signal

stopped: AtomicBool

A boolean marking if this channel is stopped

session: SessionWeakPtr

Weak pointer to respective session

version: OnceCell<Arc<VersionMessage>>

The version message of the node we are connected to. Some if the version exchange has already occurred, None otherwise.

info: ChannelInfo

Channel debug info

metering_map: Mutex<HashMap<String, MeteringQueue>>

Map holding a MeteringQueue for each [Message] to perform rate limiting of propagation towards the stream.

Implementations

impl Channel

pub async fn new( stream: Box<dyn PtStream>, resolve_addr: Option<Url>, connect_addr: Url, session: SessionWeakPtr, ) -> Arc<Self>

Sets up a new channel. Creates a reader and writer PtStream and the message publisher subsystem. Performs a network handshake on the subsystem dispatchers.

async fn setup_dispatchers(subsystem: &MessageSubsystem)

Perform network handshake for message subsystem dispatchers.

pub fn start(self: Arc<Self>, executor: Arc<Executor<'_>>)

Starts the channel. Runs a receive loop to start receiving messages or handles a network failure.

pub async fn stop(&self)

Stops the channel. Notifies all publishers that the channel has been closed in handle_stop().

pub async fn subscribe_stop(&self) -> Result<Subscription<Error>>

Creates a subscription to a stopped signal. If the channel is stopped then this will return a ChannelStopped error.

pub fn is_stopped(&self) -> bool

pub async fn send<M: Message>(&self, message: &M) -> Result<()>

Sends a message across a channel. First it converts the message into a SerializedMessage and then calls send_serialized to send it. Returns an error if something goes wrong.

pub async fn send_serialized( &self, message: &SerializedMessage, metering_score: &u64, metering_config: &MeteringConfiguration, ) -> Result<()>

Sends the encoded payload of provided SerializedMessage across the channel.

We first check if we should apply some throttling, based on the provided Message configuration. We always sleep 2x times more that the exepted one, so we don’t flood the peer. Then, calls send_message that creates a new payload and sends it over the network transport as a packet. Returns an error if something goes wrong.

async fn send_message(&self, message: &SerializedMessage) -> Result<()>

Sends the encoded payload of provided SerializedMessage by writing the data to the channel async stream.

pub async fn read_command<R: AsyncRead + Unpin + Send + Sized>( &self, stream: &mut R, ) -> Result<String>

Returns a decoded Message command. We start by extracting the length from the stream, then allocate the precise buffer for this length using stream.take(). This manual deserialization provides a basic DDOS protection, since it prevents nodes from sending an arbitarily large payload.

pub async fn subscribe_msg<M: Message>(&self) -> Result<MessageSubscription<M>>

Subscribe to a message on the message subsystem.

async fn handle_stop(self: Arc<Self>, result: Result<()>)

Handle network errors. Panic if error passes silently, otherwise broadcast the error.

async fn main_receive_loop(self: Arc<Self>) -> Result<()>

Run the receive loop. Start receiving messages or handle network failure.

pub async fn ban(&self)

Ban a malicious peer and stop the channel.

pub fn address(&self) -> &Url

Returns the relevant socket address for this connection. If this is an outbound connection, the transport-processed resolve_addr will be returned. Otherwise for inbound connections it will default to connect_addr.

pub fn resolve_addr(&self) -> Option<Url>

Returns the socket address that has undergone transport processing, if it exists. Returns None otherwise.

pub fn connect_addr(&self) -> &Url

Return the socket address without transport processing.

pub(crate) async fn set_version(&self, version: Arc<VersionMessage>)

Set the VersionMessage of the node this channel is connected to. Called on receiving a version message in ProtocolVersion.

pub fn get_version(&self) -> Arc<VersionMessage>

Should only be called after the version exchange has been completed.

pub fn message_subsystem(&self) -> &MessageSubsystem

Returns the inner MessageSubsystem reference

fn session(&self) -> Arc<dyn Session>

pub fn session_type_id(&self) -> SessionBitFlag

pub fn p2p(&self) -> P2pPtr

pub fn hosts(&self) -> HostsPtr

fn is_eof_error(err: &Error) -> bool

Trait Implementations

impl Debug for Channel

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.