feat(socketio/socket): use reserve rather than reserve_many and Iterator for Permtis

Author: Totodore

engineioxide/src/socket.rs

@@ -146,29 +146,6 @@ impl Permit<'_> {
     }
 }
 
-/// An [`Iterator`] over the permits returned by the [`reserve`](Socket::reserve) function
-#[derive(Debug)]
-pub struct PermitIterator<'a> {
-    inner: mpsc::PermitIterator<'a, PacketBuf>,
-}
-
-impl<'a> Iterator for PermitIterator<'a> {
-    type Item = Permit<'a>;
-
-    #[inline]
-    fn next(&mut self) -> Option<Self::Item> {
-        let inner = self.inner.next()?;
-        Some(Permit { inner })
-    }
-}
-impl ExactSizeIterator for PermitIterator<'_> {
-    #[inline]
-    fn len(&self) -> usize {
-        self.inner.len()
-    }
-}
-impl std::iter::FusedIterator for PermitIterator<'_> {}
-
 /// Buffered packets to send to the client
 pub(crate) type PacketBuf = SmallVec<[Packet; 10]>;
 /// A [`Socket`] represents a client connection to the server.
@@ -417,9 +394,9 @@ where
     /// If the internal chan is full, the function will return a [`TrySendError::Full`] error.
     /// If the socket is closed, the function will return a [`TrySendError::Closed`] error.
     #[inline]
-    pub fn reserve(&self, n: usize) -> Result<PermitIterator<'_>, TrySendError<()>> {
-        let inner = self.internal_tx.try_reserve_many(n)?;
-        Ok(PermitIterator { inner })
+    pub fn reserve(&self) -> Result<Permit<'_>, TrySendError<()>> {
+        let permit = self.internal_tx.try_reserve()?;
+        Ok(Permit { inner: permit })
     }
 
     /// Emits a message to the client.

socketioxide/src/handler/extract.rs

@@ -298,10 +298,10 @@ impl<A: Adapter> AckSender<A> {
 
     /// Send the ack response to the client.
     pub fn send<T: Serialize>(self, data: T) -> Result<(), SendError<T>> {
-        use crate::socket::PermitIteratorExt;
+        use crate::socket::PermitExt;
         if let Some(ack_id) = self.ack_id {
-            let permits = match self.socket.reserve(1) {
-                Ok(permits) => permits,
+            let permit = match self.socket.reserve() {
+                Ok(permit) => permit,
                 Err(e) => {
                     #[cfg(feature = "tracing")]
                     tracing::debug!("sending error during emit message: {e:?}");
@@ -315,7 +315,7 @@ impl<A: Adapter> AckSender<A> {
             } else {
                 Packet::bin_ack(ns, data, self.binary, ack_id)
             };
-            permits.emit(packet);
+            permit.send(packet);
             Ok(())
         } else {
             Ok(())

socketioxide/src/operators.rs

@@ -346,20 +346,20 @@ impl<A: Adapter> ConfOperators<'_, A> {
         data: T,
     ) -> Result<(), SendError<T>> {
         use crate::errors::SocketError;
-        use crate::socket::PermitIteratorExt;
+        use crate::socket::PermitExt;
         if !self.socket.connected() {
             return Err(SendError::Socket(SocketError::Closed(data)));
         }
-        let permits = match self.socket.reserve(1) {
-            Ok(permits) => permits,
+        let permit = match self.socket.reserve() {
+            Ok(permit) => permit,
             Err(e) => {
                 #[cfg(feature = "tracing")]
                 tracing::debug!("sending error during emit message: {e:?}");
                 return Err(e.with_value(data).into());
             }
         };
         let packet = self.get_packet(event, data)?;
-        permits.emit(packet);
+        permit.send(packet);
 
         Ok(())
     }
@@ -424,8 +424,8 @@ impl<A: Adapter> ConfOperators<'_, A> {
         if !self.socket.connected() {
             return Err(SendError::Socket(SocketError::Closed(data)));
         }
-        let permits = match self.socket.reserve(1) {
-            Ok(permits) => permits,
+        let permit = match self.socket.reserve() {
+            Ok(permit) => permit,
             Err(e) => {
                 #[cfg(feature = "tracing")]
                 tracing::debug!("sending error during emit message: {e:?}");
@@ -434,7 +434,7 @@ impl<A: Adapter> ConfOperators<'_, A> {
         };
         let timeout = self.timeout.unwrap_or(self.socket.config.ack_timeout);
         let packet = self.get_packet(event, data)?;
-        let rx = self.socket.send_with_ack_permit(packet, permits);
+        let rx = self.socket.send_with_ack_permit(packet, permit);
         let stream = AckInnerStream::send(rx, timeout, self.socket.id);
         Ok(AckStream::<V>::from(stream))
     }

socketioxide/src/socket.rs

@@ -12,7 +12,7 @@ use std::{
     time::Duration,
 };
 
-use engineioxide::socket::{DisconnectReason as EIoDisconnectReason, Permit, PermitIterator};
+use engineioxide::socket::{DisconnectReason as EIoDisconnectReason, Permit};
 use serde::{de::DeserializeOwned, Serialize};
 use serde_json::Value;
 use tokio::sync::oneshot::{self, Receiver};
@@ -101,12 +101,11 @@ impl From<EIoDisconnectReason> for DisconnectReason {
     }
 }
 
-pub(crate) trait PermitIteratorExt<'a>:
-    ExactSizeIterator<Item = Permit<'a>> + Sized
-{
-    fn emit(mut self, mut packet: Packet<'_>) {
-        debug_assert!(self.len() > 0, "No permits available to send the message");
-
+pub(crate) trait PermitExt<'a> {
+    fn send(self, packet: Packet<'_>);
+}
+impl<'a> PermitExt<'a> for Permit<'a> {
+    fn send(self, mut packet: Packet<'_>) {
         let bin_payloads = match packet.inner {
             PacketData::BinaryEvent(_, ref mut bin, _) | PacketData::BinaryAck(ref mut bin, _) => {
                 Some(std::mem::take(&mut bin.bin))
@@ -117,13 +116,12 @@ pub(crate) trait PermitIteratorExt<'a>:
         let msg = packet.into();
 
         if let Some(bin_payloads) = bin_payloads {
-            self.next().unwrap().emit_many(msg, bin_payloads);
+            self.emit_many(msg, bin_payloads);
         } else {
-            self.next().unwrap().emit(msg);
+            self.emit(msg);
         }
     }
 }
-impl<'a> PermitIteratorExt<'a> for PermitIterator<'a> {}
 
 /// A Socket represents a client connected to a namespace.
 /// It is used to send and receive messages from the client, join and leave rooms, etc.
@@ -314,8 +312,8 @@ impl<A: Adapter> Socket<A> {
             return Err(SendError::Socket(SocketError::Closed(data)));
         }
 
-        let permits = match self.reserve(1) {
-            Ok(permits) => permits,
+        let permit = match self.reserve() {
+            Ok(permit) => permit,
             Err(e) => {
                 #[cfg(feature = "tracing")]
                 tracing::debug!("sending error during emit message: {e:?}");
@@ -325,7 +323,7 @@ impl<A: Adapter> Socket<A> {
 
         let ns = self.ns();
         let data = serde_json::to_value(data)?;
-        permits.emit(Packet::event(ns, event.into(), data));
+        permit.send(Packet::event(ns, event.into(), data));
         Ok(())
     }
 
@@ -388,8 +386,8 @@ impl<A: Adapter> Socket<A> {
         if !self.connected() {
             return Err(SendError::Socket(SocketError::Closed(data)));
         }
-        let permits = match self.reserve(1) {
-            Ok(permits) => permits,
+        let permit = match self.reserve() {
+            Ok(permit) => permit,
             Err(e) => {
                 #[cfg(feature = "tracing")]
                 tracing::debug!("sending error during emit message: {e:?}");
@@ -398,7 +396,7 @@ impl<A: Adapter> Socket<A> {
         };
         let data = serde_json::to_value(data)?;
         let packet = Packet::event(self.ns(), event.into(), data);
-        let rx = self.send_with_ack_permit(packet, permits);
+        let rx = self.send_with_ack_permit(packet, permit);
         let stream = AckInnerStream::send(rx, self.config.ack_timeout, self.id);
         Ok(AckStream::<V>::from(stream))
     }
@@ -649,26 +647,26 @@ impl<A: Adapter> Socket<A> {
         &self.ns.path
     }
 
-    pub(crate) fn reserve(&self, n: usize) -> Result<PermitIterator<'_>, SocketError<()>> {
-        Ok(self.esocket.reserve(n)?)
+    pub(crate) fn reserve(&self) -> Result<Permit<'_>, SocketError<()>> {
+        Ok(self.esocket.reserve()?)
     }
 
     pub(crate) fn send(&self, packet: Packet<'_>) -> Result<(), SocketError<()>> {
-        let permits = self.reserve(1)?;
-        permits.emit(packet);
+        let permit = self.reserve()?;
+        permit.send(packet);
         Ok(())
     }
 
     pub(crate) fn send_with_ack_permit(
         &self,
         mut packet: Packet<'_>,
-        permits: PermitIterator<'_>,
+        permit: Permit<'_>,
     ) -> Receiver<AckResult<Value>> {
         let (tx, rx) = oneshot::channel();
 
         let ack = self.ack_counter.fetch_add(1, Ordering::SeqCst) + 1;
         packet.inner.set_ack_id(ack);
-        permits.emit(packet);
+        permit.send(packet);
         self.ack_message.lock().unwrap().insert(ack, tx);
         rx
     }