fn start_writer_task<T: tokio::io::AsyncWrite + Send + Unpin + 'static>(
        executor: &Handle,
        connection_metadata: ConnectionMetadata,
        network_context: Arc<NetworkContext>,
        mut writer: FramedWrite<T, LengthDelimitedCodec>,
    ) -> (
        channel::Sender<(
            NetworkMessage,
            oneshot::Sender<Result<(), PeerManagerError>>,
        )>,
        oneshot::Sender<()>,
    ) {
        let remote_peer_id = connection_metadata.remote_peer_id;
        let (write_reqs_tx, mut write_reqs_rx): (
            channel::Sender<(
                NetworkMessage,
                oneshot::Sender<Result<(), PeerManagerError>>,
            )>,
            _,
        ) = channel::new(1024, &counters::PENDING_WIRE_MESSAGES);
        let (close_tx, close_rx) = oneshot::channel();
        let writer_task = async move {
            let mut close_rx = close_rx.into_stream();
            loop {
                futures::select! {
                    (message, ack_ch) = write_reqs_rx.select_next_some() => {
                        if let Err(e) = writer
                            .send(
                                lcs::to_bytes(&message)
                                    .expect("Outbound message failed to serialize")
                                    .into(),
                            )
                            .map_ok(|_| ack_ch.send(Ok(())))
                            .await
                        {
                            warn!(
                                NetworkSchema::new(&network_context)
                                    .connection_metadata(&connection_metadata),
                                error = ?e,
                                "{} Error in sending message to peer: {}. Error: {:?}",
                                network_context,
                                remote_peer_id.short_str(),
                                e
                            );
                            break;
                        }
                    },
                    _ = close_rx.select_next_some() => {
                        break;
                    }
                }
            }
            info!(
                NetworkSchema::new(&network_context).connection_metadata(&connection_metadata),
                "{} Closing connection to peer: {}",
                network_context,
                remote_peer_id.short_str()
            );
            let flush_and_close = async move {
                writer.flush().await?;
                writer.close().await?;
                Ok(()) as io::Result<()>
            };
            match tokio::time::timeout(transport::TRANSPORT_TIMEOUT, flush_and_close).await {
                Err(_) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        "{} Timeout in flush/close of connection to peer: {}",
                        network_context,
                        remote_peer_id.short_str()
                    );
                }
                Ok(Err(e)) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        error = ?e,
                        "{} Failure in flush/close of connection to peer: {}. Error: {:?}",
                        network_context,
                        remote_peer_id.short_str(),
                        e
                    );
                }
                Ok(Ok(())) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        "{} Closed connection to peer: {}",
                        network_context,
                        remote_peer_id.short_str()
                    );
                }
            }
        };
        executor.spawn(writer_task);
        (write_reqs_tx, close_tx)
    }
// Start a new task on the given executor which is responsible for writing outbound messages on
// the wire. The function returns two channels which can be used to send intructions to the
// task:
// 1. The first channel is used to send outbound NetworkMessages to the task
// 2. The second channel is used to instruct the task to close the connection and terminate.
// If outbound messages are queued when the task receives a close instruction, it discards
// them and immediately closes the connection.
Start a new task on the given executor which is responsible for writing outbound messages on
the wire. The function returns two channels which can be used to send intructions to the
task:
1. The first channel is used to send outbound NetworkMessages to the task
2. The second channel is used to instruct the task to close the connection and terminate.
If outbound messages are queued when the task receives a close instruction, it discards
them and immediately closes the connection.
Start a new task on the given executor which is responsible for writing outbound messages on the wire . The function returns two channels which can be used to send intructions to the task : 1 . The first channel is used to send outbound NetworkMessages to the task 2 . The second channel is used to instruct the task to close the connection and terminate . If outbound messages are queued when the task receives a close instruction it discards them and immediately closes the connection .
fn start_writer_task<T: tokio::io::AsyncWrite + Send + Unpin + 'static>(
        executor: &Handle,
        connection_metadata: ConnectionMetadata,
        network_context: Arc<NetworkContext>,
        mut writer: FramedWrite<T, LengthDelimitedCodec>,
    ) -> (
        channel::Sender<(
            NetworkMessage,
            oneshot::Sender<Result<(), PeerManagerError>>,
        )>,
        oneshot::Sender<()>,
    ) {
        let remote_peer_id = connection_metadata.remote_peer_id;
        let (write_reqs_tx, mut write_reqs_rx): (
            channel::Sender<(
                NetworkMessage,
                oneshot::Sender<Result<(), PeerManagerError>>,
            )>,
            _,
        ) = channel::new(1024, &counters::PENDING_WIRE_MESSAGES);
        let (close_tx, close_rx) = oneshot::channel();
        let writer_task = async move {
            let mut close_rx = close_rx.into_stream();
            loop {
                futures::select! {
                    (message, ack_ch) = write_reqs_rx.select_next_some() => {
                        if let Err(e) = writer
                            .send(
                                lcs::to_bytes(&message)
                                    .expect("Outbound message failed to serialize")
                                    .into(),
                            )
                            .map_ok(|_| ack_ch.send(Ok(())))
                            .await
                        {
                            warn!(
                                NetworkSchema::new(&network_context)
                                    .connection_metadata(&connection_metadata),
                                error = ?e,
                                "{} Error in sending message to peer: {}. Error: {:?}",
                                network_context,
                                remote_peer_id.short_str(),
                                e
                            );
                            break;
                        }
                    },
                    _ = close_rx.select_next_some() => {
                        break;
                    }
                }
            }
            info!(
                NetworkSchema::new(&network_context).connection_metadata(&connection_metadata),
                "{} Closing connection to peer: {}",
                network_context,
                remote_peer_id.short_str()
            );
            let flush_and_close = async move {
                writer.flush().await?;
                writer.close().await?;
                Ok(()) as io::Result<()>
            };
            match tokio::time::timeout(transport::TRANSPORT_TIMEOUT, flush_and_close).await {
                Err(_) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        "{} Timeout in flush/close of connection to peer: {}",
                        network_context,
                        remote_peer_id.short_str()
                    );
                }
                Ok(Err(e)) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        error = ?e,
                        "{} Failure in flush/close of connection to peer: {}. Error: {:?}",
                        network_context,
                        remote_peer_id.short_str(),
                        e
                    );
                }
                Ok(Ok(())) => {
                    info!(
                        NetworkSchema::new(&network_context)
                            .connection_metadata(&connection_metadata),
                        "{} Closed connection to peer: {}",
                        network_context,
                        remote_peer_id.short_str()
                    );
                }
            }
        };
        executor.spawn(writer_task);
        (write_reqs_tx, close_tx)
    }
fn start_writer_task < T : tokio :: io :: AsyncWrite + Send + Unpin + ' static > ( executor : & Handle , connection_metadata : ConnectionMetadata , network_context : Arc < NetworkContext > , mut writer : FramedWrite < T , LengthDelimitedCodec > , ) -> ( channel :: Sender < ( NetworkMessage , oneshot :: Sender < Result < ( ) , PeerManagerError > > , ) > , oneshot :: Sender < ( ) > , ) { let remote_peer_id = connection_metadata . remote_peer_id ; let ( write_reqs_tx , mut write_reqs_rx ) : ( channel :: Sender < ( NetworkMessage , oneshot :: Sender < Result < ( ) , PeerManagerError > > , ) > , _ , ) = channel :: new ( 1024 , & counters :: PENDING_WIRE_MESSAGES ) ; let ( close_tx , close_rx ) = oneshot :: channel ( ) ; let writer_task = async move { let mut close_rx = close_rx . into_stream ( ) ; loop { futures :: select ! { ( message ack_ch ) write_reqs_rx select_next_some ( ) { if let Err ( e ) writer send ( lcs to_bytes ( message ) expect ( " " ) into ( ) ) map_ok ( ack_ch send ( Ok ( ( ) ) ) ) await { warn ( NetworkSchema new ( network_context ) connection_metadata ( connection_metadata ) error e " " network_context remote_peer_id short_str ( ) e ) break } } close_rx select_next_some ( ) { break } } } info ! ( NetworkSchema new ( network_context ) connection_metadata ( connection_metadata ) " " network_context remote_peer_id short_str ( ) ) ; let flush_and_close = async move { writer . flush ( ) . await ? ; writer . close ( ) . await ? ; Ok ( ( ) ) as io :: Result < ( ) > } ; match tokio :: time :: timeout ( transport :: TRANSPORT_TIMEOUT , flush_and_close ) . await { Err ( _ ) => { info ! ( NetworkSchema new ( network_context ) connection_metadata ( connection_metadata ) " " network_context remote_peer_id short_str ( ) ) ; } Ok ( Err ( e ) ) => { info ! ( NetworkSchema new ( network_context ) connection_metadata ( connection_metadata ) error e " " network_context remote_peer_id short_str ( ) e ) ; } Ok ( Ok ( ( ) ) ) => { info ! ( NetworkSchema new ( network_context ) connection_metadata ( connection_metadata ) " " network_context remote_peer_id short_str ( ) ) ; } } } ; executor . spawn ( writer_task ) ; ( write_reqs_tx , close_tx ) }
Start a new task on the given executor which is responsible for writing outbound messages on
the wire.
Start a new task on the given executor which is responsible for writing outbound messages on the wire .
executor
&Handle
connection_metadata
ConnectionMetadata
network_context
Arc<NetworkContext>
writer
FramedWrite<T, LengthDelimitedCodec>
executor
&Handle
connection_metadata
ConnectionMetadata
network_context
Arc<NetworkContext>
writer
FramedWrite<T, LengthDelimitedCodec>