pub fn initialize_with_details<T, U>(
        transport: Transport<T, U>,
        details: Option<SessionDetails>,
    ) -> io::Result<Self>
    where
        T: DataStream,
        U: Codec + Send + 'static,
    {
        let (mut t_read, mut t_write) = transport.into_split();
        let post_office = Arc::new(Mutex::new(PostOffice::new()));
        let weak_post_office = Arc::downgrade(&post_office);

        // Start a task that continually checks for responses and delivers them using the
        // post office
        let response_task = tokio::spawn(async move {
            loop {
                match t_read.receive::<Response>().await {
                    Ok(Some(res)) => {
                        trace!("Incoming response: {:?}", res);
                        let res_id = res.id;
                        let res_origin_id = res.origin_id;

                        // Try to send response to appropriate mailbox
                        // NOTE: We don't log failures as errors as using fire(...) for a
                        //       session is valid and would not have a mailbox
                        if !post_office.lock().await.deliver(res).await {
                            trace!(
                                "Response {} has no mailbox for origin {}",
                                res_id,
                                res_origin_id
                            );
                        }
                    }
                    Ok(None) => {
                        debug!("Session closing response task as transport read-half closed!");
                        break;
                    }
                    Err(x) => {
                        error!("Failed to receive response from server: {}", x);
                        break;
                    }
                }
            }

            // Clean up remaining mailbox senders
            post_office.lock().await.clear_mailboxes();
        });

        let (tx, mut rx) = mpsc::channel::<Request>(1);
        let request_task = tokio::spawn(async move {
            while let Some(req) = rx.recv().await {
                if let Err(x) = t_write.send(req).await {
                    error!("Failed to send request to server: {}", x);
                    break;
                }
            }
        });

        // Create a task that runs once a minute and prunes mailboxes
        let post_office = Weak::clone(&weak_post_office);
        let prune_task = tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(60)).await;
                if let Some(post_office) = Weak::upgrade(&post_office) {
                    post_office.lock().await.prune_mailboxes();
                } else {
                    break;
                }
            }
        });

        let channel = SessionChannel {
            tx,
            post_office: weak_post_office,
        };

        Ok(Self {
            channel,
            details,
            request_task,
            response_task,
            prune_task,
        })
    }
/// Initializes a session using the provided transport and extra details
Initializes a session using the provided transport and extra details
Initializes a session using the provided transport and extra details
pub fn initialize_with_details<T, U>(
        transport: Transport<T, U>,
        details: Option<SessionDetails>,
    ) -> io::Result<Self>
    where
        T: DataStream,
        U: Codec + Send + 'static,
    {
        let (mut t_read, mut t_write) = transport.into_split();
        let post_office = Arc::new(Mutex::new(PostOffice::new()));
        let weak_post_office = Arc::downgrade(&post_office);
        // Start a task that continually checks for responses and delivers them using the
        // post office
        let response_task = tokio::spawn(async move {
            loop {
                match t_read.receive::<Response>().await {
                    Ok(Some(res)) => {
                        trace!("Incoming response: {:?}", res);
                        let res_id = res.id;
                        let res_origin_id = res.origin_id;
                        // Try to send response to appropriate mailbox
                        // NOTE: We don't log failures as errors as using fire(...) for a
                        //       session is valid and would not have a mailbox
                        if !post_office.lock().await.deliver(res).await {
                            trace!(
                                "Response {} has no mailbox for origin {}",
                                res_id,
                                res_origin_id
                            );
                        }
                    }
                    Ok(None) => {
                        debug!("Session closing response task as transport read-half closed!");
                        break;
                    }
                    Err(x) => {
                        error!("Failed to receive response from server: {}", x);
                        break;
                    }
                }
            }
            // Clean up remaining mailbox senders
            post_office.lock().await.clear_mailboxes();
        });
        let (tx, mut rx) = mpsc::channel::<Request>(1);
        let request_task = tokio::spawn(async move {
            while let Some(req) = rx.recv().await {
                if let Err(x) = t_write.send(req).await {
                    error!("Failed to send request to server: {}", x);
                    break;
                }
            }
        });
        // Create a task that runs once a minute and prunes mailboxes
        let post_office = Weak::clone(&weak_post_office);
        let prune_task = tokio::spawn(async move {
            loop {
                tokio::time::sleep(Duration::from_secs(60)).await;
                if let Some(post_office) = Weak::upgrade(&post_office) {
                    post_office.lock().await.prune_mailboxes();
                } else {
                    break;
                }
            }
        });
        let channel = SessionChannel {
            tx,
            post_office: weak_post_office,
        };
        Ok(Self {
            channel,
            details,
            request_task,
            response_task,
            prune_task,
        })
    }
pub fn initialize_with_details < T , U > ( transport : Transport < T , U > , details : Option < SessionDetails > , ) -> io :: Result < Self > where T : DataStream , U : Codec + Send + ' static , { let ( mut t_read , mut t_write ) = transport . into_split ( ) ; let post_office = Arc :: new ( Mutex :: new ( PostOffice :: new ( ) ) ) ; let weak_post_office = Arc :: downgrade ( & post_office ) ; // Start a task that continually checks for responses and delivers them using the // post office let response_task = tokio :: spawn ( async move { loop { match t_read . receive :: < Response > ( ) . await { Ok ( Some ( res ) ) => { trace ! ( " " res ) ; let res_id = res . id ; let res_origin_id = res . origin_id ; // Try to send response to appropriate mailbox // NOTE: We don't log failures as errors as using fire(...) for a //       session is valid and would not have a mailbox if ! post_office . lock ( ) . await . deliver ( res ) . await { trace ! ( " " res_id res_origin_id ) ; } } Ok ( None ) => { debug ! ( " " ) ; break ; } Err ( x ) => { error ! ( " " x ) ; break ; } } } // Clean up remaining mailbox senders post_office . lock ( ) . await . clear_mailboxes ( ) ; } ) ; let ( tx , mut rx ) = mpsc :: channel :: < Request > ( 1 ) ; let request_task = tokio :: spawn ( async move { while let Some ( req ) = rx . recv ( ) . await { if let Err ( x ) = t_write . send ( req ) . await { error ! ( " " x ) ; break ; } } } ) ; // Create a task that runs once a minute and prunes mailboxes let post_office = Weak :: clone ( & weak_post_office ) ; let prune_task = tokio :: spawn ( async move { loop { tokio :: time :: sleep ( Duration :: from_secs ( 60 ) ) . await ; if let Some ( post_office ) = Weak :: upgrade ( & post_office ) { post_office . lock ( ) . await . prune_mailboxes ( ) ; } else { break ; } } } ) ; let channel = SessionChannel { tx , post_office : weak_post_office , } ; Ok ( Self { channel , details , request_task , response_task , prune_task , } ) }
Initializes a session using the provided transport and extra details
Initializes a session using the provided transport and extra details
transport
Transport<T, U>
details
Option<SessionDetails>