pub async fn recv(&mut self) -> Result<BytesMut, Error> {
        loop {
            let mut message = match read_message(&mut self.read).await {
                Ok(message) => message,
                Err(err) => {
                    error!("Terminating server because of: {:?}", err);
                    self.bad = true;
                    return Err(err);
                }
            };

            // Buffer the message we'll forward to the client later.
            self.buffer.put(&message[..]);

            let code = message.get_u8() as char;
            let _len = message.get_i32();

            trace!("Message: {}", code);

            match code {
                // ReadyForQuery
                'Z' => {
                    let transaction_state = message.get_u8() as char;

                    match transaction_state {
                        // In transaction.
                        'T' => {
                            self.in_transaction = true;
                        }

                        // Idle, transaction over.
                        'I' => {
                            self.in_transaction = false;
                        }

                        // Some error occured, the transaction was rolled back.
                        'E' => {
                            self.in_transaction = true;
                        }

                        // Something totally unexpected, this is not a Postgres server we know.
                        _ => {
                            self.bad = true;
                            return Err(Error::ProtocolSyncError);
                        }
                    };

                    // There is no more data available from the server.
                    self.data_available = false;

                    break;
                }

                // DataRow
                'D' => {
                    // More data is available after this message, this is not the end of the reply.
                    self.data_available = true;

                    // Don't flush yet, the more we buffer, the faster this goes...up to a limit.
                    if self.buffer.len() >= 8196 {
                        break;
                    }
                }

                // CopyInResponse: copy is starting from client to server.
                'G' => break,

                // CopyOutResponse: copy is starting from the server to the client.
                'H' => {
                    self.data_available = true;
                    break;
                }

                // CopyData: we are not buffering this one because there will be many more
                // and we don't know how big this packet could be, best not to take a risk.
                'd' => break,

                // CopyDone
                // Buffer until ReadyForQuery shows up, so don't exit the loop yet.
                'c' => (),

                // Anything else, e.g. errors, notices, etc.
                // Keep buffering until ReadyForQuery shows up.
                _ => (),
            };
        }

        let bytes = self.buffer.clone();

        // Keep track of how much data we got from the server for stats.
        self.stats
            .data_received(bytes.len(), self.process_id, self.address.id);

        // Clear the buffer for next query.
        self.buffer.clear();

        // Pass the data back to the client.
        Ok(bytes)
    }
/// Receive data from the server in response to a client request.
/// This method must be called multiple times while `self.is_data_available()` is true
/// in order to receive all data the server has to offer.
Receive data from the server in response to a client request.
This method must be called multiple times while `self.is_data_available()` is true
in order to receive all data the server has to offer.
Receive data from the server in response to a client request . This method must be called multiple times while ` self . is_data_available () ` is true in order to receive all data the server has to offer .
pub async fn recv(&mut self) -> Result<BytesMut, Error> {
        loop {
            let mut message = match read_message(&mut self.read).await {
                Ok(message) => message,
                Err(err) => {
                    error!("Terminating server because of: {:?}", err);
                    self.bad = true;
                    return Err(err);
                }
            };
            // Buffer the message we'll forward to the client later.
            self.buffer.put(&message[..]);
            let code = message.get_u8() as char;
            let _len = message.get_i32();
            trace!("Message: {}", code);
            match code {
                // ReadyForQuery
                'Z' => {
                    let transaction_state = message.get_u8() as char;
                    match transaction_state {
                        // In transaction.
                        'T' => {
                            self.in_transaction = true;
                        }
                        // Idle, transaction over.
                        'I' => {
                            self.in_transaction = false;
                        }
                        // Some error occured, the transaction was rolled back.
                        'E' => {
                            self.in_transaction = true;
                        }
                        // Something totally unexpected, this is not a Postgres server we know.
                        _ => {
                            self.bad = true;
                            return Err(Error::ProtocolSyncError);
                        }
                    };
                    // There is no more data available from the server.
                    self.data_available = false;
                    break;
                }
                // DataRow
                'D' => {
                    // More data is available after this message, this is not the end of the reply.
                    self.data_available = true;
                    // Don't flush yet, the more we buffer, the faster this goes...up to a limit.
                    if self.buffer.len() >= 8196 {
                        break;
                    }
                }
                // CopyInResponse: copy is starting from client to server.
                'G' => break,
                // CopyOutResponse: copy is starting from the server to the client.
                'H' => {
                    self.data_available = true;
                    break;
                }
                // CopyData: we are not buffering this one because there will be many more
                // and we don't know how big this packet could be, best not to take a risk.
                'd' => break,
                // CopyDone
                // Buffer until ReadyForQuery shows up, so don't exit the loop yet.
                'c' => (),
                // Anything else, e.g. errors, notices, etc.
                // Keep buffering until ReadyForQuery shows up.
                _ => (),
            };
        }
        let bytes = self.buffer.clone();
        // Keep track of how much data we got from the server for stats.
        self.stats
            .data_received(bytes.len(), self.process_id, self.address.id);
        // Clear the buffer for next query.
        self.buffer.clear();
        // Pass the data back to the client.
        Ok(bytes)
    }
pub async fn recv ( & mut self ) -> Result < BytesMut , Error > { loop { let mut message = match read_message ( & mut self . read ) . await { Ok ( message ) => message , Err ( err ) => { error ! ( " " err ) ; self . bad = true ; return Err ( err ) ; } } ; // Buffer the message we'll forward to the client later. self . buffer . put ( & message [ .. ] ) ; let code = message . get_u8 ( ) as char ; let _len = message . get_i32 ( ) ; trace ! ( " " code ) ; match code { // ReadyForQuery 'Z' => { let transaction_state = message . get_u8 ( ) as char ; match transaction_state { // In transaction. 'T' => { self . in_transaction = true ; } // Idle, transaction over. 'I' => { self . in_transaction = false ; } // Some error occured, the transaction was rolled back. 'E' => { self . in_transaction = true ; } // Something totally unexpected, this is not a Postgres server we know. _ => { self . bad = true ; return Err ( Error :: ProtocolSyncError ) ; } } ; // There is no more data available from the server. self . data_available = false ; break ; } // DataRow 'D' => { // More data is available after this message, this is not the end of the reply. self . data_available = true ; // Don't flush yet, the more we buffer, the faster this goes...up to a limit. if self . buffer . len ( ) >= 8196 { break ; } } // CopyInResponse: copy is starting from client to server. 'G' => break , // CopyOutResponse: copy is starting from the server to the client. 'H' => { self . data_available = true ; break ; } // CopyData: we are not buffering this one because there will be many more // and we don't know how big this packet could be, best not to take a risk. 'd' => break , // CopyDone // Buffer until ReadyForQuery shows up, so don't exit the loop yet. 'c' => ( ) , // Anything else, e.g. errors, notices, etc. // Keep buffering until ReadyForQuery shows up. _ => ( ) , } ; } let bytes = self . buffer . clone ( ) ; // Keep track of how much data we got from the server for stats. self . stats . data_received ( bytes . len ( ) , self . process_id , self . address . id ) ; // Clear the buffer for next query. self . buffer . clear ( ) ; // Pass the data back to the client. Ok ( bytes ) }
Receive data from the server in response to a client request.
Receive data from the server in response to a client request .
&mut self
&mut self