/// Publish message to given topic
///
/// Publishes message with the topic and QoS-class. Only Qos-classes 0 and 1 are implemented now
///
/// # Example
/// ```
/// let data = vec![1,2,3,4,5,6,7,8,9,10];
///
/// let result = amqtt.publish("test/topic", Qos::AtLeastOnce, &data).await;
/// ```
Publish message to given topic
Publishes message with the topic and QoS-class. Only Qos-classes 0 and 1 are implemented now
Publish message to given topic Publishes message with the topic and QoS - class . Only Qos - classes 0 and 1 are implemented now
pub async fn publish(
        &mut self,
        topic_name: &str,
        qos: Qos,
        payload: &[u8],
    ) -> Result<&Self, MqttError> {
        info!("--Publish--");
        // Fixed header 2 bytes + variable header fixed part + 4 bytes
        //let mut raw_buf = Vec::with_capacity(topic_name.len()+payload.len()+1);
        let mut raw_buf = [0 as u8; 4096];
        match qos {
            Qos::AtMostOnce => {
                debug!("starting Qos zero publishing to {topic_name}");
                // Encode an MQTT Connect packet.
                let pkt = Publish {
                    dup: false,
                    qospid: mqttrs::QosPid::AtMostOnce,
                    retain: false,
                    topic_name,
                    payload,
                }
                .into();
                let length = mqttrs::encode_slice(&pkt, &mut raw_buf)?;
                let buf = BytesMut::from(&raw_buf[..length]);
                let msg = ChannelPacket::Generic(MqttMsg::new(-1, buf));
                self.tx_stream
                    .send(msg)
                    .await
                    .map_err(|_| MqttError::AsyncChannelSendError)?;
            }
            Qos::AtLeastOnce | Qos::ExactlyOnce => {
                assert!(Qos::AtLeastOnce == qos);
                debug!("starting Qos non-zero publishing to {topic_name}");
                let used_pid = self.pid;
                self.pid = self.pid + 1;
                let pkt = mqttrs::Publish {
                    dup: false,
                    qospid: mqttrs::QosPid::AtLeastOnce(used_pid),
                    retain: false,
                    topic_name,
                    payload,
                }
                .into();
                let length = mqttrs::encode_slice(&pkt, &mut raw_buf)?;
                let buf = BytesMut::from(&raw_buf[..length]);
                // Get the integer version
                let pid = used_pid.get();
                let msg = MqttMsg::new((pid as u16).into(), buf);
                let fut = msg.get_packet_future();
                let channel_packet = ChannelPacket::Generic(msg);
                self.tx_stream
                    .send(channel_packet)
                    .await
                    .map_err(|_| MqttError::AsyncChannelSendError)?;
                fut.or(async {
                    for retry_count in 1..4 {
                        Timer::after(Duration::from_secs(1)).await;
                        let pkt = mqttrs::Publish {
                            dup: true, // This is resent
                            qospid: mqttrs::QosPid::AtLeastOnce(used_pid),
                            retain: false,
                            topic_name,
                            payload,
                        }
                        .into();
                        let length = mqttrs::encode_slice(&pkt, &mut raw_buf)?;
                        let buf = BytesMut::from(&raw_buf[..length]);
                        let channel_packet = ChannelPacket::Generic(MqttMsg::new(-1, buf));
                        self.tx_stream
                            .send(channel_packet)
                            .await
                            .map_err(|_| MqttError::AsyncChannelSendError)?;
                        warn!("Publish timeout and retry number {retry_count}");
                    }
                    // All retries have been exhausted if we end up this far
                    Err(MqttError::ConnectionTimeout)
                })
                .await?;
            }
        }
        Ok(self)
    }
pub async fn publish ( & mut self , topic_name : & str , qos : Qos , payload : & [ u8 ] , ) -> Result < & Self , MqttError > { info ! ( " " ) ; // Fixed header 2 bytes + variable header fixed part + 4 bytes //let mut raw_buf = Vec::with_capacity(topic_name.len()+payload.len()+1); let mut raw_buf = [ 0 as u8 ; 4096 ] ; match qos { Qos :: AtMostOnce => { debug ! ( " " ) ; // Encode an MQTT Connect packet. let pkt = Publish { dup : false , qospid : mqttrs :: QosPid :: AtMostOnce , retain : false , topic_name , payload , } . into ( ) ; let length = mqttrs :: encode_slice ( & pkt , & mut raw_buf ) ? ; let buf = BytesMut :: from ( & raw_buf [ .. length ] ) ; let msg = ChannelPacket :: Generic ( MqttMsg :: new ( - 1 , buf ) ) ; self . tx_stream . send ( msg ) . await . map_err ( | _ | MqttError :: AsyncChannelSendError ) ? ; } Qos :: AtLeastOnce | Qos :: ExactlyOnce => { assert ! ( Qos AtLeastOnce qos ) ; debug ! ( " " ) ; let used_pid = self . pid ; self . pid = self . pid + 1 ; let pkt = mqttrs :: Publish { dup : false , qospid : mqttrs :: QosPid :: AtLeastOnce ( used_pid ) , retain : false , topic_name , payload , } . into ( ) ; let length = mqttrs :: encode_slice ( & pkt , & mut raw_buf ) ? ; let buf = BytesMut :: from ( & raw_buf [ .. length ] ) ; // Get the integer version let pid = used_pid . get ( ) ; let msg = MqttMsg :: new ( ( pid as u16 ) . into ( ) , buf ) ; let fut = msg . get_packet_future ( ) ; let channel_packet = ChannelPacket :: Generic ( msg ) ; self . tx_stream . send ( channel_packet ) . await . map_err ( | _ | MqttError :: AsyncChannelSendError ) ? ; fut . or ( async { for retry_count in 1 .. 4 { Timer :: after ( Duration :: from_secs ( 1 ) ) . await ; let pkt = mqttrs :: Publish { dup : true , // This is resent qospid : mqttrs :: QosPid :: AtLeastOnce ( used_pid ) , retain : false , topic_name , payload , } . into ( ) ; let length = mqttrs :: encode_slice ( & pkt , & mut raw_buf ) ? ; let buf = BytesMut :: from ( & raw_buf [ .. length ] ) ; let channel_packet = ChannelPacket :: Generic ( MqttMsg :: new ( - 1 , buf ) ) ; self . tx_stream . send ( channel_packet ) . await . map_err ( | _ | MqttError :: AsyncChannelSendError ) ? ; warn ! ( " " ) ; } // All retries have been exhausted if we end up this far Err ( MqttError :: ConnectionTimeout ) } ) . await ? ; } } Ok ( self ) }
Publish message to given topic
Publishes message with the topic and QoS-class.
Publish message to given topic Publishes message with the topic and QoS - class .
&mut self
topic_name
&str
qos
Qos
payload
&[u8]
example


let result = amqtt.publish("test/topic", Qos::AtLeastOnce, &data).await;
```
let result = amqtt . publish ( " test / topic " Qos :: AtLeastOnce &data ) . await ; ` ` `