pub fn poll<F>(&mut self, mut f: F) -> Result<(), Error<TcpStack::Error>>
    where
        for<'a> F: FnMut(
            &mut MqttClient<TcpStack, Clock, MSG_SIZE, MSG_COUNT>,
            &'a str,
            &[u8],
            &[Property<'a>],
        ),
    {
        self.client.process()?;

        // If the connection is no longer active, reset the packet reader state and return. There's
        // nothing more we can do.
        if self.client.connection_state.state() != &States::Active
            && self.client.connection_state.state() != &States::Establishing
        {
            self.packet_reader.reset();
            return Ok(());
        }

        let mut buf: [u8; 1024] = [0; 1024];
        let received = self.client.network.read(&mut buf)?;
        if received > 0 {
            debug!("Received {} bytes", received);
        }

        let mut processed = 0;
        while processed < received {
            match self.packet_reader.slurp(&buf[processed..received]) {
                Ok(count) => {
                    debug!("Processed {} bytes", count);
                    processed += count
                }

                Err(e) => {
                    self.client.reset();
                    self.packet_reader.reset();
                    return Err(Error::Protocol(e));
                }
            }

            // Handle any received packets.
            while self.packet_reader.packet_available() {
                let packet = ReceivedPacket::parse_message(&self.packet_reader)?;

                info!("Received {:?}", packet);

                let result = self.client.handle_packet(packet, &mut f);

                self.packet_reader.pop_packet()?;

                // If there was an error, return it now. Note that we ensure the packet is removed
                // from buffering after processing even in error conditions..
                result?;
            }
        }

        Ok(())
    }
/// Check the MQTT interface for available messages.
///
/// # Args
/// * `f` - A closure to process any received messages. The closure should accept the client,
/// topic, message, and list of proprties (in that order).
Check the MQTT interface for available messages.
Check the MQTT interface for available messages .
pub fn poll<F>(&mut self, mut f: F) -> Result<(), Error<TcpStack::Error>>
    where
        for<'a> F: FnMut(
            &mut MqttClient<TcpStack, Clock, MSG_SIZE, MSG_COUNT>,
            &'a str,
            &[u8],
            &[Property<'a>],
        ),
    {
        self.client.process()?;
        // If the connection is no longer active, reset the packet reader state and return. There's
        // nothing more we can do.
        if self.client.connection_state.state() != &States::Active
            && self.client.connection_state.state() != &States::Establishing
        {
            self.packet_reader.reset();
            return Ok(());
        }
        let mut buf: [u8; 1024] = [0; 1024];
        let received = self.client.network.read(&mut buf)?;
        if received > 0 {
            debug!("Received {} bytes", received);
        }
        let mut processed = 0;
        while processed < received {
            match self.packet_reader.slurp(&buf[processed..received]) {
                Ok(count) => {
                    debug!("Processed {} bytes", count);
                    processed += count
                }
                Err(e) => {
                    self.client.reset();
                    self.packet_reader.reset();
                    return Err(Error::Protocol(e));
                }
            }
            // Handle any received packets.
            while self.packet_reader.packet_available() {
                let packet = ReceivedPacket::parse_message(&self.packet_reader)?;
                info!("Received {:?}", packet);
                let result = self.client.handle_packet(packet, &mut f);
                self.packet_reader.pop_packet()?;
                // If there was an error, return it now. Note that we ensure the packet is removed
                // from buffering after processing even in error conditions..
                result?;
            }
        }
        Ok(())
    }
pub fn poll < F > ( & mut self , mut f : F ) -> Result < ( ) , Error < TcpStack :: Error > > where for < ' a > F : FnMut ( & mut MqttClient < TcpStack , Clock , MSG_SIZE , MSG_COUNT > , & ' a str , & [ u8 ] , & [ Property < ' a > ] , ) , { self . client . process ( ) ? ; // If the connection is no longer active, reset the packet reader state and return. There's // nothing more we can do. if self . client . connection_state . state ( ) != & States :: Active && self . client . connection_state . state ( ) != & States :: Establishing { self . packet_reader . reset ( ) ; return Ok ( ( ) ) ; } let mut buf : [ u8 ; 1024 ] = [ 0 ; 1024 ] ; let received = self . client . network . read ( & mut buf ) ? ; if received > 0 { debug ! ( " " received ) ; } let mut processed = 0 ; while processed < received { match self . packet_reader . slurp ( & buf [ processed .. received ] ) { Ok ( count ) => { debug ! ( " " count ) ; processed += count } Err ( e ) => { self . client . reset ( ) ; self . packet_reader . reset ( ) ; return Err ( Error :: Protocol ( e ) ) ; } } // Handle any received packets. while self . packet_reader . packet_available ( ) { let packet = ReceivedPacket :: parse_message ( & self . packet_reader ) ? ; info ! ( " " packet ) ; let result = self . client . handle_packet ( packet , & mut f ) ; self . packet_reader . pop_packet ( ) ? ; // If there was an error, return it now. Note that we ensure the packet is removed // from buffering after processing even in error conditions.. result ? ; } } Ok ( ( ) ) }
Check the MQTT interface for available messages.
Check the MQTT interface for available messages .
&mut self
f
F
args
`f` - A closure to process any received messages. The closure should accept the client,
topic, message, and list of proprties (in that order).
` f ` - A closure to process any received messages . The closure should accept the client topic message and list of proprties ( in that order ) .