async fn send_fetch_stream_request(&self, socket: &mut KfSocket) -> Result<(), KfSocketError> {
        let local_spu_id = self.config.id();
        trace!(
            "sending fetch stream for leader: {} for follower: {}",
            self.leader_id,
            local_spu_id
        );
        let mut fetch_request = FetchStreamRequest::default();
        fetch_request.spu_id = local_spu_id;
        let mut message = RequestMessage::new_request(fetch_request);
        message
            .get_mut_header()
            .set_client_id(format!("peer spu: {}", local_spu_id));

        let response = socket.send(&message).await?;
        trace!("fetch stream response: {:#?}", response);
        debug!(
            "established peer to peer channel to leader: {} from follower: {}",
            self.leader_id, local_spu_id
        );
        Ok(())
    }
/// send request to establish peer to peer communication to leader
send request to establish peer to peer communication to leader
send request to establish peer to peer communication to leader
async fn send_fetch_stream_request(&self, socket: &mut KfSocket) -> Result<(), KfSocketError> {
        let local_spu_id = self.config.id();
        trace!(
            "sending fetch stream for leader: {} for follower: {}",
            self.leader_id,
            local_spu_id
        );
        let mut fetch_request = FetchStreamRequest::default();
        fetch_request.spu_id = local_spu_id;
        let mut message = RequestMessage::new_request(fetch_request);
        message
            .get_mut_header()
            .set_client_id(format!("peer spu: {}", local_spu_id));
        let response = socket.send(&message).await?;
        trace!("fetch stream response: {:#?}", response);
        debug!(
            "established peer to peer channel to leader: {} from follower: {}",
            self.leader_id, local_spu_id
        );
        Ok(())
    }
async fn send_fetch_stream_request ( & self , socket : & mut KfSocket ) -> Result < ( ) , KfSocketError > { let local_spu_id = self . config . id ( ) ; trace ! ( " " self leader_id local_spu_id ) ; let mut fetch_request = FetchStreamRequest :: default ( ) ; fetch_request . spu_id = local_spu_id ; let mut message = RequestMessage :: new_request ( fetch_request ) ; message . get_mut_header ( ) . set_client_id ( format ! ( " " local_spu_id ) ) ; let response = socket . send ( & message ) . await ? ; trace ! ( " " response ) ; debug ! ( " " self leader_id local_spu_id ) ; Ok ( ( ) ) }
send request to establish peer to peer communication to leader
send request to establish peer to peer communication to leader
&self
socket
&mut KfSocket
&self
socket
&mut KfSocket