fn init_common<T: AsyncReadExt + AsyncWriteExt + Send + Unpin + 'static>(stream: T) -> Self {
        let (rx_stream, tx_stream) = stream.split();
        //let rx_stream = stream;
        //let tx_stream = rx_stream.clone();

        let (s, r) = unbounded(); // channels for IPC between receiver and client

        let mut mqtt = Mqtt::new(tx_stream);

        info!("ready_spawn_receiver");

        smol::spawn(async move {
            info!("Entering receive loop");
            mqtt.process_packets(rx_stream, r)
                .await
                .map_err(|x| {
                    warn!("Exited receive loop {:?}", x);
                })
                .unwrap_err();
        })
        .detach();

        Self {
            pid: mqttrs::Pid::new(),
            tx_stream: s,
        }
    }
/// Creates new insecure MQTT client
///
/// Creates standard non-TLS MQTT connection. MQTT TCP/IP port is 1883 but some brokers/application can choose to use something
/// different. Connect needs to be called to finalize the setup and register the client-id etc.
///
Creates new insecure MQTT client
Creates standard non-TLS MQTT connection. MQTT TCP/IP port is 1883 but some brokers/application can choose to use something
different. Connect needs to be called to finalize the setup and register the client-id etc.
Creates new insecure MQTT client Creates standard non - TLS MQTT connection . MQTT TCP / IP port is 1883 but some brokers / application can choose to use something different . Connect needs to be called to finalize the setup and register the client - id etc .
fn init_common<T: AsyncReadExt + AsyncWriteExt + Send + Unpin + 'static>(stream: T) -> Self {
        let (rx_stream, tx_stream) = stream.split();
        //let rx_stream = stream;
        //let tx_stream = rx_stream.clone();
        let (s, r) = unbounded(); // channels for IPC between receiver and client
        let mut mqtt = Mqtt::new(tx_stream);
        info!("ready_spawn_receiver");
        smol::spawn(async move {
            info!("Entering receive loop");
            mqtt.process_packets(rx_stream, r)
                .await
                .map_err(|x| {
                    warn!("Exited receive loop {:?}", x);
                })
                .unwrap_err();
        })
        .detach();
        Self {
            pid: mqttrs::Pid::new(),
            tx_stream: s,
        }
    }
fn init_common < T : AsyncReadExt + AsyncWriteExt + Send + Unpin + ' static > ( stream : T ) -> Self { let ( rx_stream , tx_stream ) = stream . split ( ) ; //let rx_stream = stream; //let tx_stream = rx_stream.clone(); let ( s , r ) = unbounded ( ) ; // channels for IPC between receiver and client let mut mqtt = Mqtt :: new ( tx_stream ) ; info ! ( " " ) ; smol :: spawn ( async move { info ! ( " " ) ; mqtt . process_packets ( rx_stream , r ) . await . map_err ( | x | { warn ! ( " " x ) ; } ) . unwrap_err ( ) ; } ) . detach ( ) ; Self { pid : mqttrs :: Pid :: new ( ) , tx_stream : s , } }
Creates new insecure MQTT client
Creates standard non-TLS MQTT connection.
Creates new insecure MQTT client Creates standard non - TLS MQTT connection .
stream
T
stream
T