pub async fn subscribe(
        &mut self,
        mqtt_topics: &Vec<(String, Qos)>,
    ) -> Result<Vec<Result<(Receiver<Vec<u8>>, Qos), MqttError>>, MqttError> {
        info!("--Subscribe--");
        // Fixed header 2 bytes + variable header fixed part 3 per topic
        let mut raw_buf: [u8; 1024] = [0; 1024];

        let topics: Vec<mqttrs::SubscribeTopic> = mqtt_topics
            .into_iter()
            .map(|(topic_path, qos)| mqttrs::SubscribeTopic {
                topic_path: topic_path.to_owned(),
                qos: match qos {
                    Qos::AtMostOnce => mqttrs::QoS::AtMostOnce,
                    Qos::AtLeastOnce => mqttrs::QoS::AtLeastOnce,
                    Qos::ExactlyOnce => mqttrs::QoS::ExactlyOnce,
                },
            })
            .collect();

        let status = self
            .send_packet(|pid| {
                // Encode an MQTT Connect packet.
                let pkt = Subscribe { pid, topics }.into();

                let length = mqttrs::encode_slice(&pkt, &mut raw_buf)?;
                let buf = BytesMut::from(&raw_buf[..length]);

                let pid = pid.get();

                let msg = MqttMsg::new((pid as u16).into(), buf);
                let fut = msg.get_packet_future();
                let channel_packet = ChannelPacket::Generic(msg);

                Ok((channel_packet, fut, pid))
            })
            .await?;

        // We need to use the inner as the outer is async
        info!("End of subscribe");

        if let PacketFutureResult::Subscriptions(subs) = status {
            Ok(subs)
        } else {
            panic!("We should never be here");
        }
    }
/// Subscribe to a set of topics.
///
/// Subscribes the client to set of topics with the defined quality of the service. The outer result
/// indicates how the overall operation performed and each topic inside has a result of its own related to the
/// status of the subscription.
///
/// The returned Receivers channels are used to communicate packets to given topic and they match the order in which the topics were supplied.
///
/// # Example
/// ```
/// let test_subscription = vec![(String::from("test/topic"), Qos::AtLeastOnce),(String::from("test/topic1"), Qos::AtMostOnce)];
///
/// let subs_result = amqtt.subscribe(&test_subscription).await;
///
/// if let Ok(subs) = &subs_result {
///     for (i, individual_sub_result) in subs.iter().enumerate() {
///         match individual_sub_result {
///             Ok((_, qos)) => info!("Subscription for {} succeeded with Qos {:?}", test_subscription[i].0, qos),
///             Err(err) => panic!("MQTT error {:?}", err)
///         }
///     }
/// }
/// ```
Subscribe to a set of topics.
Subscribes the client to set of topics with the defined quality of the service. The outer result
indicates how the overall operation performed and each topic inside has a result of its own related to the
status of the subscription.

The returned Receivers channels are used to communicate packets to given topic and they match the order in which the topics were supplied.
Subscribe to a set of topics . Subscribes the client to set of topics with the defined quality of the service . The outer result indicates how the overall operation performed and each topic inside has a result of its own related to the status of the subscription . The returned Receivers channels are used to communicate packets to given topic and they match the order in which the topics were supplied .
pub async fn subscribe(
        &mut self,
        mqtt_topics: &Vec<(String, Qos)>,
    ) -> Result<Vec<Result<(Receiver<Vec<u8>>, Qos), MqttError>>, MqttError> {
        info!("--Subscribe--");
        // Fixed header 2 bytes + variable header fixed part 3 per topic
        let mut raw_buf: [u8; 1024] = [0; 1024];
        let topics: Vec<mqttrs::SubscribeTopic> = mqtt_topics
            .into_iter()
            .map(|(topic_path, qos)| mqttrs::SubscribeTopic {
                topic_path: topic_path.to_owned(),
                qos: match qos {
                    Qos::AtMostOnce => mqttrs::QoS::AtMostOnce,
                    Qos::AtLeastOnce => mqttrs::QoS::AtLeastOnce,
                    Qos::ExactlyOnce => mqttrs::QoS::ExactlyOnce,
                },
            })
            .collect();
        let status = self
            .send_packet(|pid| {
                // Encode an MQTT Connect packet.
                let pkt = Subscribe { pid, topics }.into();
                let length = mqttrs::encode_slice(&pkt, &mut raw_buf)?;
                let buf = BytesMut::from(&raw_buf[..length]);
                let pid = pid.get();
                let msg = MqttMsg::new((pid as u16).into(), buf);
                let fut = msg.get_packet_future();
                let channel_packet = ChannelPacket::Generic(msg);
                Ok((channel_packet, fut, pid))
            })
            .await?;
        // We need to use the inner as the outer is async
        info!("End of subscribe");
        if let PacketFutureResult::Subscriptions(subs) = status {
            Ok(subs)
        } else {
            panic!("We should never be here");
        }
    }
pub async fn subscribe ( & mut self , mqtt_topics : & Vec < ( String , Qos ) > , ) -> Result < Vec < Result < ( Receiver < Vec < u8 > > , Qos ) , MqttError > > , MqttError > { info ! ( " " ) ; // Fixed header 2 bytes + variable header fixed part 3 per topic let mut raw_buf : [ u8 ; 1024 ] = [ 0 ; 1024 ] ; let topics : Vec < mqttrs :: SubscribeTopic > = mqtt_topics . into_iter ( ) . map ( | ( topic_path , qos ) | mqttrs :: SubscribeTopic { topic_path : topic_path . to_owned ( ) , qos : match qos { Qos :: AtMostOnce => mqttrs :: QoS :: AtMostOnce , Qos :: AtLeastOnce => mqttrs :: QoS :: AtLeastOnce , Qos :: ExactlyOnce => mqttrs :: QoS :: ExactlyOnce , } , } ) . collect ( ) ; let status = self . send_packet ( | pid | { // Encode an MQTT Connect packet. let pkt = Subscribe { pid , topics } . into ( ) ; let length = mqttrs :: encode_slice ( & pkt , & mut raw_buf ) ? ; let buf = BytesMut :: from ( & raw_buf [ .. length ] ) ; let pid = pid . get ( ) ; let msg = MqttMsg :: new ( ( pid as u16 ) . into ( ) , buf ) ; let fut = msg . get_packet_future ( ) ; let channel_packet = ChannelPacket :: Generic ( msg ) ; Ok ( ( channel_packet , fut , pid ) ) } ) . await ? ; // We need to use the inner as the outer is async info ! ( " " ) ; if let PacketFutureResult :: Subscriptions ( subs ) = status { Ok ( subs ) } else { panic ! ( " " ) ; } }
Subscribe to a set of topics.
Subscribe to a set of topics .
&mut self
mqtt_topics
&Vec<(String, Qos)>
&mut self
mqtt_topics
&Vec<(String, Qos)>
example
```
let test_subscription = vec![(String::from("test/topic"), Qos::AtLeastOnce),(String::from("test/topic1"), Qos::AtMostOnce)].

let subs_result = amqtt.subscribe(&test_subscription).await.

if let Ok(subs) = &subs_result {
for (i, individual_sub_result) in subs.iter().enumerate() {
match individual_sub_result {
Ok((_, qos)) => info!("Subscription for {} succeeded with Qos {:?}", test_subscription[i].0, qos),
Err(err) => panic!("MQTT error {:?}", err)
}
}
}
```
` ` ` let test_subscription = vec! [ ( String :: from ( " test / topic " ) Qos :: AtLeastOnce ) ( String :: from ( " test / topic1 " ) Qos :: AtMostOnce ) ] . let subs_result = amqtt . subscribe ( &test_subscription ) . await . if let Ok ( subs ) = &subs_result { for ( i individual_sub_result ) in subs . iter () . enumerate () { match individual_sub_result { Ok (( _ qos )) = > info! ( " Subscription for {} succeeded with Qos { : ? } " test_subscription [ i ] . 0 qos ) Err ( err ) = > panic! ( " MQTT error { : ? } " err ) } } } ` ` `