pub async fn run(&self, socket: &mut TcpStream) -> Result<(), ObjectReceiveError> {
        // save peer address for reporting/debugging purposes
        let peer_address = socket
            .peer_addr()
            .map_err(|err| ObjectReceiveError::ConnectionError(None, err))?;
        let num_objects = self.object_ids.len();
        info!("receiving {} objects from {}", num_objects, peer_address);

        // read the first byte of the response; BEGIN indicates the the peer is about to start
        // sending objects; otherwise, there was some kind of error on the peer side an nothing
        // will be sent
        let status = socket
            .read_u8()
            .await
            .map_err(|err| ObjectReceiveError::ConnectionError(Some(peer_address), err))?;
        if status != status_codes::BEGIN {
            return Err(ObjectReceiveError::PeerError(peer_address, status));
        }

        // receive objects one-by-one, and save them to the local plasma store.
        let plasma_object_ids = map_object_ids(&self.object_ids);
        let mut bytes_received = 0;
        for (i, oid) in plasma_object_ids.iter().enumerate() {
            match receive_object(&self.plasma_client, oid, socket, peer_address).await {
                Ok(ob) => {
                    debug!("received object {} from {}", ob, peer_address);
                    bytes_received += ob.size();
                }
                Err(err) => {
                    // try to return to pre-request state by deleting already received objects;
                    // if the delete fails, just swallow the error
                    let _ = self
                        .plasma_client
                        .delete_many(&plasma_object_ids[..(i + 1)]);
                    return Err(err);
                }
            };
        }

        // all objects have been received - so, remove them from the receiving set
        info!(
            "received {} objects ({} bytes) from {}",
            num_objects, bytes_received, peer_address
        );
        Ok(())
    }
/// Reads objects from the specified socket and saves them into the local plasma store;
/// the objects are assumed to be order in the order specified by `object_ids` list.
///
/// Will return an error if:
/// * The peer sends an error code as the first byte of the response.
/// * Creating and sealing an object in the local plasma store fails for any reason.
/// * Peer closes connection for any reason.
Reads objects from the specified socket and saves them into the local plasma store;
the objects are assumed to be order in the order specified by `object_ids` list.

Will return an error if:
The peer sends an error code as the first byte of the response.
Creating and sealing an object in the local plasma store fails for any reason.
Peer closes connection for any reason.
Reads objects from the specified socket and saves them into the local plasma store ; the objects are assumed to be order in the order specified by ` object_ids ` list . Will return an error if : The peer sends an error code as the first byte of the response . Creating and sealing an object in the local plasma store fails for any reason . Peer closes connection for any reason .
pub async fn run(&self, socket: &mut TcpStream) -> Result<(), ObjectReceiveError> {
        // save peer address for reporting/debugging purposes
        let peer_address = socket
            .peer_addr()
            .map_err(|err| ObjectReceiveError::ConnectionError(None, err))?;
        let num_objects = self.object_ids.len();
        info!("receiving {} objects from {}", num_objects, peer_address);
        // read the first byte of the response; BEGIN indicates the the peer is about to start
        // sending objects; otherwise, there was some kind of error on the peer side an nothing
        // will be sent
        let status = socket
            .read_u8()
            .await
            .map_err(|err| ObjectReceiveError::ConnectionError(Some(peer_address), err))?;
        if status != status_codes::BEGIN {
            return Err(ObjectReceiveError::PeerError(peer_address, status));
        }
        // receive objects one-by-one, and save them to the local plasma store.
        let plasma_object_ids = map_object_ids(&self.object_ids);
        let mut bytes_received = 0;
        for (i, oid) in plasma_object_ids.iter().enumerate() {
            match receive_object(&self.plasma_client, oid, socket, peer_address).await {
                Ok(ob) => {
                    debug!("received object {} from {}", ob, peer_address);
                    bytes_received += ob.size();
                }
                Err(err) => {
                    // try to return to pre-request state by deleting already received objects;
                    // if the delete fails, just swallow the error
                    let _ = self
                        .plasma_client
                        .delete_many(&plasma_object_ids[..(i + 1)]);
                    return Err(err);
                }
            };
        }
        // all objects have been received - so, remove them from the receiving set
        info!(
            "received {} objects ({} bytes) from {}",
            num_objects, bytes_received, peer_address
        );
        Ok(())
    }
pub async fn run ( & self , socket : & mut TcpStream ) -> Result < ( ) , ObjectReceiveError > { // save peer address for reporting/debugging purposes let peer_address = socket . peer_addr ( ) . map_err ( | err | ObjectReceiveError :: ConnectionError ( None , err ) ) ? ; let num_objects = self . object_ids . len ( ) ; info ! ( " " num_objects peer_address ) ; // read the first byte of the response; BEGIN indicates the the peer is about to start // sending objects; otherwise, there was some kind of error on the peer side an nothing // will be sent let status = socket . read_u8 ( ) . await . map_err ( | err | ObjectReceiveError :: ConnectionError ( Some ( peer_address ) , err ) ) ? ; if status != status_codes :: BEGIN { return Err ( ObjectReceiveError :: PeerError ( peer_address , status ) ) ; } // receive objects one-by-one, and save them to the local plasma store. let plasma_object_ids = map_object_ids ( & self . object_ids ) ; let mut bytes_received = 0 ; for ( i , oid ) in plasma_object_ids . iter ( ) . enumerate ( ) { match receive_object ( & self . plasma_client , oid , socket , peer_address ) . await { Ok ( ob ) => { debug ! ( " " ob peer_address ) ; bytes_received += ob . size ( ) ; } Err ( err ) => { // try to return to pre-request state by deleting already received objects; // if the delete fails, just swallow the error let _ = self . plasma_client . delete_many ( & plasma_object_ids [ .. ( i + 1 ) ] ) ; return Err ( err ) ; } } ; } // all objects have been received - so, remove them from the receiving set info ! ( " " num_objects bytes_received peer_address ) ; Ok ( ( ) ) }
Reads objects from the specified socket and saves them into the local plasma store;
the objects are assumed to be order in the order specified by `object_ids` list.
Reads objects from the specified socket and saves them into the local plasma store ; the objects are assumed to be order in the order specified by ` object_ids ` list .
&self
socket
&mut TcpStream
&self
socket
&mut TcpStream