// Poll for incoming messages and buffer them as needed. This
// defines a bounded loop that reads up to `MESSAGE_BATCH_SIZE`
// messages every tick.
Poll for incoming messages and buffer them as needed. This
defines a bounded loop that reads up to `MESSAGE_BATCH_SIZE`
messages every tick.
Poll for incoming messages and buffer them as needed . This defines a bounded loop that reads up to ` MESSAGE_BATCH_SIZE ` messages every tick .
fn poll(&mut self) -> Poll<(), io::Error> {
        for i in 0..MESSAGE_BATCH_SIZE {
            // Polling an `UnboundedReceiver` cannot fail, so `unwrap`
            // here is safe.
            match self.rx.poll().unwrap() {
                Async::Ready(Some(v)) => {
                    // Buffer the message.
                    self.client.buffer(&v);
                    // If `MESSAGE_BATCH_SIZE` is hit, there are still
                    // messages left to read on the last iteration of the
                    // loop before it breaks. Notify the executor of this
                    // task to schedule it again.
                    if i + 1 == MESSAGE_BATCH_SIZE {
                        task::current().notify();
                    }
                }
                _ => break,
            }
        }
        // Flush the write buffer to the socket.
        let _ = self.client.poll_flush()?;
        // Read new messages from the socket.
        while let Async::Ready(message) = self.client.poll()? {
            if let Some(raw_message) = message {
                let _s = match str::from_utf8(&raw_message) {
                    Ok(v) => {
                        let entity_key = self.id.clone();
                        &self.session.lock().unwrap().handle_message(entity_key, String::from(v));
                        ()
                    }
                    Err(_e) => (),
                };
            } else {
                // If EOF was reached, the client has disconnected and there
                // is nothing left to do.
                return Ok(Async::Ready(()));
            }
        }
        // This should only return `NotReady` if an inner future also
        // returned `NotReady`. At this point we received a `NotReady`
        // from either `self.rx` or `self.client` and this future can
        // end.
        Ok(Async::NotReady)
    }
fn poll ( & mut self ) -> Poll < ( ) , io :: Error > { for i in 0 .. MESSAGE_BATCH_SIZE { // Polling an `UnboundedReceiver` cannot fail, so `unwrap` // here is safe. match self . rx . poll ( ) . unwrap ( ) { Async :: Ready ( Some ( v ) ) => { // Buffer the message. self . client . buffer ( & v ) ; // If `MESSAGE_BATCH_SIZE` is hit, there are still // messages left to read on the last iteration of the // loop before it breaks. Notify the executor of this // task to schedule it again. if i + 1 == MESSAGE_BATCH_SIZE { task :: current ( ) . notify ( ) ; } } _ => break , } } // Flush the write buffer to the socket. let _ = self . client . poll_flush ( ) ? ; // Read new messages from the socket. while let Async :: Ready ( message ) = self . client . poll ( ) ? { if let Some ( raw_message ) = message { let _s = match str :: from_utf8 ( & raw_message ) { Ok ( v ) => { let entity_key = self . id . clone ( ) ; & self . session . lock ( ) . unwrap ( ) . handle_message ( entity_key , String :: from ( v ) ) ; ( ) } Err ( _e ) => ( ) , } ; } else { // If EOF was reached, the client has disconnected and there // is nothing left to do. return Ok ( Async :: Ready ( ( ) ) ) ; } } // This should only return `NotReady` if an inner future also // returned `NotReady`. At this point we received a `NotReady` // from either `self.rx` or `self.client` and this future can // end. Ok ( Async :: NotReady ) }
Poll for incoming messages and buffer them as needed.
Poll for incoming messages and buffer them as needed .
&mut self