RFC(rumqttc): `publish` / `subscribe` / `unsubscribe` methods return a promise that resolves into pkid when packet is handled by `Eventloop`

[de-sh]

This is the RFC for API changes in publish / subscribe / unsubscribe in rumqttc to resolve #349. Feel free to comment and provide feedback, thanks!

Example in the case of Publish

// async
let promise = async_client.publish(..).await?;
let pkid = promise.await;

// blocking
let promise = client.publish(..)?;
let pkid = promise.blocking_recv();

Within rumqttc this will be implemented similar to how oneshot channels work, i.e. here we will respond onto the oneshot channel with the pkid value. Also we need not go so far when we know that the publish is on QoS 0 or when pkid is non-zero by default maybe?

Please note that this is not a breaking change as users can just ignore/drop the promise token.

[swanandx]

so something like, publish will return Pkid where:

struct Pkid {
  id_rx: Receiver<u32>,
}

impl Future for Pkid {
  type Output = u32;
  fn poll(..) -> .. {
    // if we recv something on -d_rx, return id
    // we will send on that channel for state.
  }
}

right?

so I am curious how are we going to send the Sender to state so we can use it for sending the pkid?

when pkid is non-zero by default maybe?

when is it non-zero by default?

as users can just ignore/drop the promise token.

that would mean the receiver is dropped as well. this would cause the Sender to fail!

[de-sh]

impl Future for Pkid {
 type Output = u32;
 fn poll(..) -> .. {
   // if we recv something on -d_rx, return id
   // we will send on that channel for state.
 }
}

We can just use oneshot as is, no need of reimplementing

so I am curious how are we going to send the Sender to state so we can use it for sending the pkid?

Include it in the request?

when is it non-zero by default?

When the publish is edited before being sent, or in the case of uplink when it is read back from file, user can choose to do that, Eventloop will skip the pkid set stage in outgoing_*

that would mean the receiver is dropped as well. this would cause the Sender to fail!

Not in the case of oneshot, it might error out, but we need not care about this either ways from within Eventloop

[swanandx]

Include it in the request?

gotcha. so unlike main...pkid-poc , by using the channels we are avoiding direct use of Arc<Mutex<_>> !

Not in the case of oneshot, it might error out,

that is what i meant by fail 😅

but we need not care about this either ways from within Eventloop

okie.

[dlips]

Hi, I am one of the possible users of this functionality from #349 :) Thanks for bringing this up again

Could you elaborate on the usage pattern for the returned promises in a multi-message scenario?

For a single message, it makes sense to wait for the return of the pkid and then continue. In case that I have a high channel capacity and an limited inflight queue, then the task using the client would block/wait at high load peaks, where the event loop would need some time to process the message. Sure, I could spawn a task that listens to the one-shot channel and after receiving the pkid it passes it on to an mpsc that gathers all the pkids.

Also the processing of the promise is a bit time critical because if I do it to late, the one-shot channel would still have a pkid value, whose PubAck was already processed by the eventloop.

Sorry that I don't have any constructive feedback, just dropping some thoughts from the user perspective

[de-sh]

Could you elaborate on the usage pattern for the returned promises in a multi-message scenario?

1. We could do a lot of things here when considering async code, but not much when it comes to sync code.
2. One could use the joinset.join_next() to stream pkids

[dlips]

joinset.join_next() is exactly what I was looking for. Missed it when looking through the tokio docs. That would solve my problem. Thank you :)

[de-sh]

Should this feature be feature gated to ensure we don't inadvertently make life worse for other users? I am not entirely sure how memory/cpu usage would increase, maybe a good idea to figure out with some profiling @swanandx

[cartertinney]

@de-sh Do we have a timeline for this inclusion? I need to be able to get pkid in order to wait for puback, in order to use this library.

[de-sh]

@de-sh Do we have a timeline for this inclusion? I need to be able to get pkid in order to wait for puback, in order to use this library.

We need a code review on the PR mentioned above to continue with this

[de-sh]

Adding more thoughts on this:

Maybe it is a good idea to have this same mechanism return the error responses relating to the publish packet so that the client side can keep track of the errors as well?

enum Error {
  Mqtt(mqttbytes::Error),
  PacketIdZero,
  EmptySubscription,
  OutgoingPacketTooLarge { .. },
}

type RequestResult = Result<Pkid, Error>;