Skip to content

Commit b5935fb

Browse files
mendessmendess
committed
Add ackable events
1 parent 3748c5b commit b5935fb

7 files changed

Lines changed: 348 additions & 54 deletions

File tree

‎socketio/src/asynchronous/client/builder.rs‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ use super::{
1717
client::{Client, ReconnectSettings},
1818
};
1919
use crate::asynchronous::socket::Socket as InnerSocket;
20+
use crate::AckId;
2021

2122
/// A builder class for a `socket.io` socket. This handles setting up the client and
2223
/// configuring the callback, the namespace and metadata of the socket. If no
@@ -190,6 +191,51 @@ impl ClientBuilder {
190191
+ 'static
191192
+ Send
192193
+ Sync,
194+
{
195+
self.on.insert(
196+
event.into(),
197+
Callback::<DynAsyncCallback>::new_no_ack(callback),
198+
);
199+
self
200+
}
201+
202+
/// Registers a new callback for a certain [`crate::event::Event`] that expects the client to
203+
/// ack. The event could either be one of the common events like `message`, `error`, `open`,
204+
/// `close` or a custom event defined by a string, e.g. `onPayment` or `foo`.
205+
///
206+
/// # Example
207+
/// ```rust
208+
/// use rust_socketio::{asynchronous::{ClientBuilder, Client}, AckId, Payload};
209+
/// use futures_util::FutureExt;
210+
///
211+
/// #[tokio::main]
212+
/// async fn main() {
213+
/// let socket = ClientBuilder::new("http://localhost:4200/")
214+
/// .namespace("/admin")
215+
/// .on("test", |payload: Payload, client: Client, ack: AckId| {
216+
/// async move {
217+
/// match payload {
218+
/// Payload::Text(values) => println!("Received: {:#?}", values),
219+
/// Payload::Binary(bin_data) => println!("Received bytes: {:#?}", bin_data),
220+
/// // This is deprecated, use Payload::Text instead
221+
/// Payload::String(str) => println!("Received: {}", str),
222+
/// }
223+
/// client.ack(ack, "received").await;
224+
/// }
225+
/// .boxed()
226+
/// })
227+
/// .on("error", |err, _| async move { eprintln!("Error: {:#?}", err) }.boxed())
228+
/// .connect()
229+
/// .await;
230+
/// }
231+
///
232+
#[cfg(feature = "async-callbacks")]
233+
pub fn on_with_ack<T: Into<Event>, F>(mut self, event: T, callback: F) -> Self
234+
where
235+
F: for<'a> std::ops::FnMut(Payload, Client, AckId) -> BoxFuture<'static, ()>
236+
+ 'static
237+
+ Send
238+
+ Sync,
193239
{
194240
self.on
195241
.insert(event.into(), Callback::<DynAsyncCallback>::new(callback));
@@ -257,6 +303,41 @@ impl ClientBuilder {
257303
pub fn on_any<F>(mut self, callback: F) -> Self
258304
where
259305
F: for<'a> FnMut(Event, Payload, Client) -> BoxFuture<'static, ()> + 'static + Send + Sync,
306+
{
307+
self.on_any = Some(Callback::<DynAsyncAnyCallback>::new_no_ack(callback));
308+
self
309+
}
310+
311+
/// Registers a Callback for all [`crate::event::Event::Custom`] and
312+
/// [`crate::event::Event::Message`] that expect the client to ack.
313+
///
314+
/// # Example
315+
/// ```rust
316+
/// use rust_socketio::{asynchronous::ClientBuilder, Payload};
317+
/// use futures_util::future::FutureExt;
318+
///
319+
/// #[tokio::main]
320+
/// async fn main() {
321+
/// let client = ClientBuilder::new("http://localhost:4200/")
322+
/// .namespace("/admin")
323+
/// .on_any(|event, payload, client, ack| {
324+
/// async {
325+
/// if let Payload::String(str) = payload {
326+
/// println!("{}: {}", String::from(event), str);
327+
/// }
328+
/// client.ack(ack, "received").await;
329+
/// }.boxed()
330+
/// })
331+
/// .connect()
332+
/// .await;
333+
/// }
334+
/// ```
335+
pub fn on_any_with_ack<F>(mut self, callback: F) -> Self
336+
where
337+
F: for<'a> FnMut(Event, Payload, Client, AckId) -> BoxFuture<'static, ()>
338+
+ 'static
339+
+ Send
340+
+ Sync,
260341
{
261342
self.on_any = Some(Callback::<DynAsyncAnyCallback>::new(callback));
262343
self

‎socketio/src/asynchronous/client/callback.rs‎

Lines changed: 56 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,27 @@
1-
use futures_util::future::BoxFuture;
1+
use futures_util::{future::BoxFuture, FutureExt};
22
use std::{
33
fmt::Debug,
4+
future::Future,
45
ops::{Deref, DerefMut},
56
};
67

7-
use crate::{Event, Payload};
8+
use crate::{AckId, Event, Payload};
89

910
use super::client::{Client, ReconnectSettings};
1011

1112
/// Internal type, provides a way to store futures and return them in a boxed manner.
12-
pub(crate) type DynAsyncCallback =
13-
Box<dyn for<'a> FnMut(Payload, Client) -> BoxFuture<'static, ()> + 'static + Send + Sync>;
13+
pub(crate) type DynAsyncCallback = Box<
14+
dyn for<'a> FnMut(Payload, Client, Option<AckId>) -> BoxFuture<'static, ()>
15+
+ 'static
16+
+ Send
17+
+ Sync,
18+
>;
1419

1520
pub(crate) type DynAsyncAnyCallback = Box<
16-
dyn for<'a> FnMut(Event, Payload, Client) -> BoxFuture<'static, ()> + 'static + Send + Sync,
21+
dyn for<'a> FnMut(Event, Payload, Client, Option<AckId>) -> BoxFuture<'static, ()>
22+
+ 'static
23+
+ Send
24+
+ Sync,
1725
>;
1826

1927
pub(crate) type DynAsyncReconnectSettingsCallback =
@@ -30,8 +38,10 @@ impl<T> Debug for Callback<T> {
3038
}
3139

3240
impl Deref for Callback<DynAsyncCallback> {
33-
type Target =
34-
dyn for<'a> FnMut(Payload, Client) -> BoxFuture<'static, ()> + 'static + Sync + Send;
41+
type Target = dyn for<'a> FnMut(Payload, Client, Option<AckId>) -> BoxFuture<'static, ()>
42+
+ 'static
43+
+ Sync
44+
+ Send;
3545

3646
fn deref(&self) -> &Self::Target {
3747
self.inner.as_ref()
@@ -45,19 +55,34 @@ impl DerefMut for Callback<DynAsyncCallback> {
4555
}
4656

4757
impl Callback<DynAsyncCallback> {
48-
pub(crate) fn new<T>(callback: T) -> Self
58+
pub(crate) fn new<T>(mut callback: T) -> Self
4959
where
50-
T: for<'a> FnMut(Payload, Client) -> BoxFuture<'static, ()> + 'static + Sync + Send,
60+
T: for<'a> FnMut(Payload, Client, AckId) -> BoxFuture<'static, ()> + 'static + Sync + Send,
5161
{
5262
Callback {
53-
inner: Box::new(callback),
63+
inner: Box::new(move |p, c, a| match a {
64+
Some(a) => callback(p, c, a).boxed(),
65+
None => std::future::ready(()).boxed(),
66+
}),
67+
}
68+
}
69+
70+
pub(crate) fn new_no_ack<T, Fut>(mut callback: T) -> Self
71+
where
72+
T: FnMut(Payload, Client) -> Fut + Sync + Send + 'static,
73+
Fut: Future<Output = ()> + 'static + Send,
74+
{
75+
Callback {
76+
inner: Box::new(move |p, c, _a| callback(p, c).boxed()),
5477
}
5578
}
5679
}
5780

5881
impl Deref for Callback<DynAsyncAnyCallback> {
59-
type Target =
60-
dyn for<'a> FnMut(Event, Payload, Client) -> BoxFuture<'static, ()> + 'static + Sync + Send;
82+
type Target = dyn for<'a> FnMut(Event, Payload, Client, Option<AckId>) -> BoxFuture<'static, ()>
83+
+ 'static
84+
+ Sync
85+
+ Send;
6186

6287
fn deref(&self) -> &Self::Target {
6388
self.inner.as_ref()
@@ -71,12 +96,28 @@ impl DerefMut for Callback<DynAsyncAnyCallback> {
7196
}
7297

7398
impl Callback<DynAsyncAnyCallback> {
74-
pub(crate) fn new<T>(callback: T) -> Self
99+
pub(crate) fn new<T>(mut callback: T) -> Self
75100
where
76-
T: for<'a> FnMut(Event, Payload, Client) -> BoxFuture<'static, ()> + 'static + Sync + Send,
101+
T: for<'a> FnMut(Event, Payload, Client, AckId) -> BoxFuture<'static, ()>
102+
+ 'static
103+
+ Sync
104+
+ Send,
77105
{
78106
Callback {
79-
inner: Box::new(callback),
107+
inner: Box::new(move |e, p, c, a| match a {
108+
Some(a) => callback(e, p, c, a).boxed(),
109+
None => std::future::ready(()).boxed(),
110+
}),
111+
}
112+
}
113+
114+
pub(crate) fn new_no_ack<T, Fut>(mut callback: T) -> Self
115+
where
116+
T: FnMut(Event, Payload, Client) -> Fut + Sync + Send + 'static,
117+
Fut: Future<Output = ()> + 'static + Send,
118+
{
119+
Callback {
120+
inner: Box::new(move |e, p, c, _a| callback(e, p, c).boxed()),
80121
}
81122
}
82123
}

‎socketio/src/asynchronous/client/client.rs‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -367,7 +367,7 @@ impl Client {
367367
id,
368368
time_started: Instant::now(),
369369
timeout,
370-
callback: Callback::<DynAsyncCallback>::new(callback),
370+
callback: Callback::<DynAsyncCallback>::new_no_ack(callback),
371371
};
372372

373373
// add the ack to the tuple of outstanding acks
@@ -376,19 +376,33 @@ impl Client {
376376
self.socket.read().await.send(socket_packet).await
377377
}
378378

379-
async fn callback<P: Into<Payload>>(&self, event: &Event, payload: P) -> Result<()> {
379+
pub async fn ack<D>(&self, ack_id: AckId, data: D) -> Result<()>
380+
where
381+
D: Into<Payload>,
382+
{
383+
let socket_packet = Packet::new_ack(data.into(), &self.nsp, ack_id);
384+
385+
self.socket.read().await.send(socket_packet).await
386+
}
387+
388+
async fn callback<P: Into<Payload>>(
389+
&self,
390+
event: &Event,
391+
payload: P,
392+
ack_id: Option<AckId>,
393+
) -> Result<()> {
380394
let mut builder = self.builder.write().await;
381395
let payload = payload.into();
382396

383397
if let Some(callback) = builder.on.get_mut(event) {
384-
callback(payload.clone(), self.clone()).await;
398+
callback(payload.clone(), self.clone(), ack_id).await;
385399
}
386400

387401
// Call on_any for all common and custom events.
388402
match event {
389403
Event::Message | Event::Custom(_) => {
390404
if let Some(callback) = builder.on_any.as_mut() {
391-
callback(event.clone(), payload, self.clone()).await;
405+
callback(event.clone(), payload, self.clone(), ack_id).await;
392406
}
393407
}
394408
_ => (),
@@ -411,6 +425,7 @@ impl Client {
411425
ack.callback.deref_mut()(
412426
Payload::from(payload.to_owned()),
413427
self.clone(),
428+
None,
414429
)
415430
.await;
416431
}
@@ -419,6 +434,7 @@ impl Client {
419434
ack.callback.deref_mut()(
420435
Payload::Binary(payload.to_owned()),
421436
self.clone(),
437+
None,
422438
)
423439
.await;
424440
}
@@ -446,8 +462,12 @@ impl Client {
446462

447463
if let Some(attachments) = &packet.attachments {
448464
if let Some(binary_payload) = attachments.get(0) {
449-
self.callback(&event, Payload::Binary(binary_payload.to_owned()))
450-
.await?;
465+
self.callback(
466+
&event,
467+
Payload::Binary(binary_payload.to_owned()),
468+
packet.id,
469+
)
470+
.await?;
451471
}
452472
}
453473
Ok(())
@@ -480,7 +500,7 @@ impl Client {
480500
};
481501

482502
// call the correct callback
483-
self.callback(&event, payloads.to_vec()).await?;
503+
self.callback(&event, payloads.to_vec(), packet.id).await?;
484504
}
485505

486506
Ok(())
@@ -495,22 +515,22 @@ impl Client {
495515
match packet.packet_type {
496516
PacketId::Ack | PacketId::BinaryAck => {
497517
if let Err(err) = self.handle_ack(packet).await {
498-
self.callback(&Event::Error, err.to_string()).await?;
518+
self.callback(&Event::Error, err.to_string(), None).await?;
499519
return Err(err);
500520
}
501521
}
502522
PacketId::BinaryEvent => {
503523
if let Err(err) = self.handle_binary_event(packet).await {
504-
self.callback(&Event::Error, err.to_string()).await?;
524+
self.callback(&Event::Error, err.to_string(), None).await?;
505525
}
506526
}
507527
PacketId::Connect => {
508528
*(self.disconnect_reason.write().await) = DisconnectReason::default();
509-
self.callback(&Event::Connect, "").await?;
529+
self.callback(&Event::Connect, "", None).await?;
510530
}
511531
PacketId::Disconnect => {
512532
*(self.disconnect_reason.write().await) = DisconnectReason::Server;
513-
self.callback(&Event::Close, "").await?;
533+
self.callback(&Event::Close, "", None).await?;
514534
}
515535
PacketId::ConnectError => {
516536
self.callback(
@@ -520,12 +540,13 @@ impl Client {
520540
.data
521541
.as_ref()
522542
.unwrap_or(&String::from("\"No error message provided\"")),
543+
None,
523544
)
524545
.await?;
525546
}
526547
PacketId::Event => {
527548
if let Err(err) = self.handle_event(packet).await {
528-
self.callback(&Event::Error, err.to_string()).await?;
549+
self.callback(&Event::Error, err.to_string(), None).await?;
529550
}
530551
}
531552
}
@@ -547,7 +568,7 @@ impl Client {
547568
None => None,
548569
Some(Err(err)) => {
549570
// call the error callback
550-
match self.callback(&Event::Error, err.to_string()).await {
571+
match self.callback(&Event::Error, err.to_string(), None).await {
551572
Err(callback_err) => Some((Err(callback_err), socket)),
552573
Ok(_) => Some((Err(err), socket)),
553574
}

0 commit comments

Comments
 (0)