About the realization of Qos 2

[StanLi-ops]

Whether the implementation of QoS 2 is currently considered. Or can it be achieved by calling send_at_least_once()?

[fafhrd91]

I don't need qos2, so I never tried to implement it. but PR is welcome.

[StanLi-ops]

Hello @fafhrd91
I am going to implement QoS2 for a project. Currently I am learning the relevant code of QoS1 in the project.
In send_at_least_once() I didn't find any related operation about "DUP".

[fafhrd91]

what do you mean "dup" related operation?

ntex-mqtt/src/v3/sink.rs

Line 175 in 09d4930

pub fn dup(mut self, val: bool) -> Self {

[StanLi-ops]

I tried to implement "step 4" by calling "send_at_least_once()", but all failed.

[StanLi-ops]

let me think, this seems more complicated than i thought

This is the qos1 I'm currently trying to implement, I'm not sure it's the right way to do it.

async fn publish_v5(
    session: v5::Session<MySession>,
    publish: v5::Publish,
) -> Result<v5::PublishAck, MyServerError> {
    log::info!(
        "incoming publish ({:?}) : {:?} -> {:?}",
        session.state(),
        publish.id(),
        publish.topic()
    );

    let sink = session.sink();

    let builder = sink.publish("test", Bytes::from("test"));
    let interval = ntex::time::interval(core::time::Duration::from_secs(1));

    let send_at_least_once = futures::FutureExt::fuse(builder.send_at_least_once());
    let timeout = futures::FutureExt::fuse(interval.tick());

    futures::pin_mut!(send_at_least_once, timeout);

    loop {
        futures::select! {
           _ = send_at_least_once  => {
               println!("send_at_least_once");
               break
            },
            _ = timeout => println!("timeout"),
        }
    }

    Ok(publish.ack())
}

[fafhrd91]

problem is in ntex-mqtt, in case of timeout ack handler wont be removed so could stall or fail for different ack. ntex needs some changes

[StanLi-ops]

problem is in ntex-mqtt, in case of timeout ack handler wont be removed so could stall or fail for different ack. ntex needs some changes

qos2 has two retransmission mechanisms similar to qos1, so in order to ensure the same usage as qos1 as much as possible, I hope this problem can be solved as soon as possible.

[StanLi-ops]

problem is in ntex-mqtt, in case of timeout ack handler wont be removed so could stall or fail for different ack. ntex needs some changes

I read the code of v5::sink::send_at_least_once_inner(). My understanding is that each publish has a corresponding "publish ack channel" to process the returned ack, so I don't quite understand the meaning of the sentence "could stall or fail for different ack". Can you explain in detail?