Skip to content

Commit 919c981

Browse files
de-shde-sh
authored
fix: token interfaces (#946)
1 parent 8be8afe commit 919c981

File tree

11 files changed

+272
-261
lines changed

11 files changed

+272
-261
lines changed

‎rumqttc/examples/ack_promise.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
3131
.unwrap()
3232
.await
3333
{
34-
Ok(pkid) => println!("Acknowledged Sub({pkid})"),
34+
Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
3535
Err(e) => println!("Subscription failed: {e:?}"),
3636
}
3737

@@ -46,7 +46,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
4646
.unwrap()
4747
.await
4848
{
49-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
49+
Ok(ack) => println!("Acknowledged Pub({ack:?})"),
5050
Err(e) => println!("Publish failed: {e:?}"),
5151
}
5252
}
@@ -66,14 +66,14 @@ async fn main() -> Result<(), Box<dyn Error>> {
6666

6767
while let Some(Ok(res)) = set.join_next().await {
6868
match res {
69-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
69+
Ok(ack) => println!("Acknowledged Pub({ack:?})"),
7070
Err(e) => println!("Publish failed: {e:?}"),
7171
}
7272
}
7373

7474
// Unsubscribe and wait for broker acknowledgement
7575
match client.unsubscribe("hello/world").await.unwrap().await {
76-
Ok(pkid) => println!("Acknowledged Unsub({pkid})"),
76+
Ok(ack) => println!("Acknowledged Unsub({ack:?})"),
7777
Err(e) => println!("Unsubscription failed: {e:?}"),
7878
}
7979

‎rumqttc/examples/ack_promise_sync.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ fn main() -> Result<(), Box<dyn Error>> {
2828
.unwrap()
2929
.wait()
3030
{
31-
Ok(pkid) => println!("Acknowledged Sub({pkid})"),
31+
Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
3232
Err(e) => println!("Subscription failed: {e:?}"),
3333
}
3434

@@ -42,7 +42,7 @@ fn main() -> Result<(), Box<dyn Error>> {
4242
.unwrap()
4343
.wait()
4444
{
45-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
45+
Ok(ack) => println!("Acknowledged Pub({ack:?})"),
4646
Err(e) => println!("Publish failed: {e:?}"),
4747
}
4848
}
@@ -83,14 +83,14 @@ fn main() -> Result<(), Box<dyn Error>> {
8383

8484
while let Ok(res) = rx.recv() {
8585
match res {
86-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
86+
Ok(ack) => println!("Acknowledged Pub({ack:?})"),
8787
Err(e) => println!("Publish failed: {e:?}"),
8888
}
8989
}
9090

9191
// Unsubscribe and wait for broker acknowledgement
9292
match client.unsubscribe("hello/world").unwrap().wait() {
93-
Ok(pkid) => println!("Acknowledged Unsub({pkid})"),
93+
Ok(ack) => println!("Acknowledged Unsub({ack:?})"),
9494
Err(e) => println!("Unsubscription failed: {e:?}"),
9595
}
9696

‎rumqttc/examples/ack_promise_v5.rs

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
3131
.unwrap()
3232
.await
3333
{
34-
Ok(pkid) => println!("Acknowledged Sub({pkid})"),
34+
Ok(pkid) => println!("Acknowledged Sub({pkid:?})"),
3535
Err(e) => println!("Subscription failed: {e:?}"),
3636
}
3737

@@ -46,7 +46,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
4646
.unwrap()
4747
.await
4848
{
49-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
49+
Ok(pkid) => println!("Acknowledged Pub({pkid:?})"),
5050
Err(e) => println!("Publish failed: {e:?}"),
5151
}
5252
}
@@ -66,14 +66,14 @@ async fn main() -> Result<(), Box<dyn Error>> {
6666

6767
while let Some(Ok(res)) = set.join_next().await {
6868
match res {
69-
Ok(pkid) => println!("Acknowledged Pub({pkid})"),
69+
Ok(pkid) => println!("Acknowledged Pub({pkid:?})"),
7070
Err(e) => println!("Publish failed: {e:?}"),
7171
}
7272
}
7373

7474
// Unsubscribe and wait for broker acknowledgement
7575
match client.unsubscribe("hello/world").await.unwrap().await {
76-
Ok(pkid) => println!("Acknowledged Unsub({pkid})"),
76+
Ok(pkid) => println!("Acknowledged Unsub({pkid:?})"),
7777
Err(e) => println!("Unsubscription failed: {e:?}"),
7878
}
7979

‎rumqttc/src/client.rs

Lines changed: 34 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,8 @@ use std::time::Duration;
55
use crate::mqttbytes::{v4::*, QoS};
66
use crate::tokens::{NoResponse, Resolver, Token};
77
use crate::{
8-
valid_filter, valid_topic, ConnectionError, Event, EventLoop, MqttOptions, Pkid, Request,
8+
valid_filter, valid_topic, AckOfAck, AckOfPub, ConnectionError, Event, EventLoop, MqttOptions,
9+
Request,
910
};
1011

1112
use bytes::Bytes;
@@ -75,7 +76,7 @@ impl AsyncClient {
7576
qos: QoS,
7677
retain: bool,
7778
payload: V,
78-
) -> Result<Token<Pkid>, ClientError>
79+
) -> Result<Token<AckOfPub>, ClientError>
7980
where
8081
S: Into<String>,
8182
V: Into<Vec<u8>>,
@@ -100,7 +101,7 @@ impl AsyncClient {
100101
qos: QoS,
101102
retain: bool,
102103
payload: V,
103-
) -> Result<Token<Pkid>, ClientError>
104+
) -> Result<Token<AckOfPub>, ClientError>
104105
where
105106
S: Into<String>,
106107
V: Into<Vec<u8>>,
@@ -119,7 +120,7 @@ impl AsyncClient {
119120
}
120121

121122
/// Sends a MQTT PubAck to the `EventLoop`. Only needed in if `manual_acks` flag is set.
122-
pub async fn ack(&self, publish: &Publish) -> Result<Token<NoResponse>, ClientError> {
123+
pub async fn ack(&self, publish: &Publish) -> Result<Token<AckOfAck>, ClientError> {
123124
let (resolver, token) = Resolver::new();
124125
let ack = get_ack_req(publish, resolver);
125126
if let Some(ack) = ack {
@@ -130,7 +131,7 @@ impl AsyncClient {
130131
}
131132

132133
/// Attempts to send a MQTT PubAck to the `EventLoop`. Only needed in if `manual_acks` flag is set.
133-
pub fn try_ack(&self, publish: &Publish) -> Result<Token<NoResponse>, ClientError> {
134+
pub fn try_ack(&self, publish: &Publish) -> Result<Token<AckOfAck>, ClientError> {
134135
let (resolver, token) = Resolver::new();
135136
let ack = get_ack_req(publish, resolver);
136137
if let Some(ack) = ack {
@@ -147,7 +148,7 @@ impl AsyncClient {
147148
qos: QoS,
148149
retain: bool,
149150
payload: Bytes,
150-
) -> Result<Token<Pkid>, ClientError>
151+
) -> Result<Token<AckOfPub>, ClientError>
151152
where
152153
S: Into<String>,
153154
{
@@ -165,7 +166,7 @@ impl AsyncClient {
165166
&self,
166167
topic: S,
167168
qos: QoS,
168-
) -> Result<Token<Pkid>, ClientError> {
169+
) -> Result<Token<SubAck>, ClientError> {
169170
let (resolver, token) = Resolver::new();
170171
let subscribe = Subscribe::new(topic, qos);
171172

@@ -184,7 +185,7 @@ impl AsyncClient {
184185
&self,
185186
topic: S,
186187
qos: QoS,
187-
) -> Result<Token<Pkid>, ClientError> {
188+
) -> Result<Token<SubAck>, ClientError> {
188189
let (resolver, token) = Resolver::new();
189190
let subscribe = Subscribe::new(topic, qos);
190191
let is_valid = subscribe_has_valid_filters(&subscribe);
@@ -198,7 +199,7 @@ impl AsyncClient {
198199
}
199200

200201
/// Sends a MQTT Subscribe for multiple topics to the `EventLoop`
201-
pub async fn subscribe_many<T>(&self, topics: T) -> Result<Token<Pkid>, ClientError>
202+
pub async fn subscribe_many<T>(&self, topics: T) -> Result<Token<SubAck>, ClientError>
202203
where
203204
T: IntoIterator<Item = SubscribeFilter>,
204205
{
@@ -216,7 +217,7 @@ impl AsyncClient {
216217
}
217218

218219
/// Attempts to send a MQTT Subscribe for multiple topics to the `EventLoop`
219-
pub fn try_subscribe_many<T>(&self, topics: T) -> Result<Token<Pkid>, ClientError>
220+
pub fn try_subscribe_many<T>(&self, topics: T) -> Result<Token<SubAck>, ClientError>
220221
where
221222
T: IntoIterator<Item = SubscribeFilter>,
222223
{
@@ -233,7 +234,10 @@ impl AsyncClient {
233234
}
234235

235236
/// Sends a MQTT Unsubscribe to the `EventLoop`
236-
pub async fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<Token<Pkid>, ClientError> {
237+
pub async fn unsubscribe<S: Into<String>>(
238+
&self,
239+
topic: S,
240+
) -> Result<Token<UnsubAck>, ClientError> {
237241
let (resolver, token) = Resolver::new();
238242
let unsubscribe = Unsubscribe::new(topic.into());
239243
let request = Request::Unsubscribe(unsubscribe, resolver);
@@ -243,7 +247,10 @@ impl AsyncClient {
243247
}
244248

245249
/// Attempts to send a MQTT Unsubscribe to the `EventLoop`
246-
pub fn try_unsubscribe<S: Into<String>>(&self, topic: S) -> Result<Token<Pkid>, ClientError> {
250+
pub fn try_unsubscribe<S: Into<String>>(
251+
&self,
252+
topic: S,
253+
) -> Result<Token<UnsubAck>, ClientError> {
247254
let (resolver, token) = Resolver::new();
248255
let unsubscribe = Unsubscribe::new(topic.into());
249256
let request = Request::Unsubscribe(unsubscribe, resolver);
@@ -271,10 +278,10 @@ impl AsyncClient {
271278
}
272279
}
273280

274-
fn get_ack_req(publish: &Publish, resolver: Resolver<()>) -> Option<Request> {
281+
fn get_ack_req(publish: &Publish, resolver: Resolver<AckOfAck>) -> Option<Request> {
275282
let ack = match publish.qos {
276283
QoS::AtMostOnce => {
277-
resolver.resolve(());
284+
resolver.resolve(AckOfAck::None);
278285
return None;
279286
}
280287
QoS::AtLeastOnce => Request::PubAck(PubAck::new(publish.pkid), resolver),
@@ -331,7 +338,7 @@ impl Client {
331338
qos: QoS,
332339
retain: bool,
333340
payload: V,
334-
) -> Result<Token<Pkid>, ClientError>
341+
) -> Result<Token<AckOfPub>, ClientError>
335342
where
336343
S: Into<String>,
337344
V: Into<Vec<u8>>,
@@ -355,7 +362,7 @@ impl Client {
355362
qos: QoS,
356363
retain: bool,
357364
payload: V,
358-
) -> Result<Token<Pkid>, ClientError>
365+
) -> Result<Token<AckOfPub>, ClientError>
359366
where
360367
S: Into<String>,
361368
V: Into<Vec<u8>>,
@@ -364,7 +371,7 @@ impl Client {
364371
}
365372

366373
/// Sends a MQTT PubAck to the `EventLoop`. Only needed in if `manual_acks` flag is set.
367-
pub fn ack(&self, publish: &Publish) -> Result<Token<NoResponse>, ClientError> {
374+
pub fn ack(&self, publish: &Publish) -> Result<Token<AckOfAck>, ClientError> {
368375
let (resolver, token) = Resolver::new();
369376
let ack = get_ack_req(publish, resolver);
370377
if let Some(ack) = ack {
@@ -375,7 +382,7 @@ impl Client {
375382
}
376383

377384
/// Sends a MQTT PubAck to the `EventLoop`. Only needed in if `manual_acks` flag is set.
378-
pub fn try_ack(&self, publish: &Publish) -> Result<Token<NoResponse>, ClientError> {
385+
pub fn try_ack(&self, publish: &Publish) -> Result<Token<AckOfAck>, ClientError> {
379386
self.client.try_ack(publish)
380387
}
381388

@@ -384,7 +391,7 @@ impl Client {
384391
&self,
385392
topic: S,
386393
qos: QoS,
387-
) -> Result<Token<Pkid>, ClientError> {
394+
) -> Result<Token<SubAck>, ClientError> {
388395
let (resolver, token) = Resolver::new();
389396
let subscribe = Subscribe::new(topic, qos);
390397
let is_valid = subscribe_has_valid_filters(&subscribe);
@@ -402,12 +409,12 @@ impl Client {
402409
&self,
403410
topic: S,
404411
qos: QoS,
405-
) -> Result<Token<Pkid>, ClientError> {
412+
) -> Result<Token<SubAck>, ClientError> {
406413
self.client.try_subscribe(topic, qos)
407414
}
408415

409416
/// Sends a MQTT Subscribe for multiple topics to the `EventLoop`
410-
pub fn subscribe_many<T>(&self, topics: T) -> Result<Token<Pkid>, ClientError>
417+
pub fn subscribe_many<T>(&self, topics: T) -> Result<Token<SubAck>, ClientError>
411418
where
412419
T: IntoIterator<Item = SubscribeFilter>,
413420
{
@@ -424,15 +431,15 @@ impl Client {
424431
Ok(token)
425432
}
426433

427-
pub fn try_subscribe_many<T>(&self, topics: T) -> Result<Token<Pkid>, ClientError>
434+
pub fn try_subscribe_many<T>(&self, topics: T) -> Result<Token<SubAck>, ClientError>
428435
where
429436
T: IntoIterator<Item = SubscribeFilter>,
430437
{
431438
self.client.try_subscribe_many(topics)
432439
}
433440

434441
/// Sends a MQTT Unsubscribe to the `EventLoop`
435-
pub fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<Token<Pkid>, ClientError> {
442+
pub fn unsubscribe<S: Into<String>>(&self, topic: S) -> Result<Token<UnsubAck>, ClientError> {
436443
let (resolver, token) = Resolver::new();
437444
let unsubscribe = Unsubscribe::new(topic.into());
438445
let request = Request::Unsubscribe(unsubscribe, resolver);
@@ -442,7 +449,10 @@ impl Client {
442449
}
443450

444451
/// Sends a MQTT Unsubscribe to the `EventLoop`
445-
pub fn try_unsubscribe<S: Into<String>>(&self, topic: S) -> Result<Token<Pkid>, ClientError> {
452+
pub fn try_unsubscribe<S: Into<String>>(
453+
&self,
454+
topic: S,
455+
) -> Result<Token<UnsubAck>, ClientError> {
446456
self.client.try_unsubscribe(topic)
447457
}
448458

‎rumqttc/src/lib.rs

Lines changed: 21 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,21 @@ pub use proxy::{Proxy, ProxyAuth, ProxyType};
160160

161161
pub type Incoming = Packet;
162162

163+
/// Used to encapsulate all publish/pubrec acknowledgements in v4
164+
#[derive(Debug, PartialEq)]
165+
pub enum AckOfPub {
166+
PubAck(PubAck),
167+
PubComp(PubComp),
168+
None,
169+
}
170+
171+
/// Used to encapsulate all ack/pubrel acknowledgements in v4
172+
#[derive(Debug)]
173+
pub enum AckOfAck {
174+
None,
175+
PubRel(PubRel),
176+
}
177+
163178
/// Current outgoing activity on the eventloop
164179
#[derive(Debug, Clone, PartialEq, Eq)]
165180
pub enum Outgoing {
@@ -191,12 +206,12 @@ pub enum Outgoing {
191206
/// handled one by one.
192207
#[derive(Debug)]
193208
pub enum Request {
194-
Publish(Publish, Resolver<Pkid>),
195-
PubAck(PubAck, Resolver<()>),
196-
PubRec(PubRec, Resolver<()>),
197-
PubRel(PubRel, Resolver<Pkid>),
198-
Subscribe(Subscribe, Resolver<Pkid>),
199-
Unsubscribe(Unsubscribe, Resolver<Pkid>),
209+
Publish(Publish, Resolver<AckOfPub>),
210+
PubAck(PubAck, Resolver<AckOfAck>),
211+
PubRec(PubRec, Resolver<AckOfAck>),
212+
PubRel(PubRel, Resolver<AckOfPub>),
213+
Subscribe(Subscribe, Resolver<SubAck>),
214+
Unsubscribe(Unsubscribe, Resolver<UnsubAck>),
200215
Disconnect(Resolver<()>),
201216
PingReq,
202217
}

0 commit comments

Comments
 (0)