feat: expose ConnectionId newtype (#133)

This will allow us to expose a `fn stop(SubscriptionId)` on the Client.

Author: obmarg

src/client/actor.rs

@@ -9,7 +9,7 @@ use serde_json::{json, Value};
 use crate::{
     logging::{trace, warning},
     protocol::Event,
-    Error,
+    Error, SubscriptionId,
 };
 
 use super::{
@@ -27,8 +27,8 @@ use super::{
 pub struct ConnectionActor {
     client: async_channel::Receiver<ConnectionCommand>,
     connection: Box<dyn ObjectSafeConnection>,
-    dropped_ids: async_channel::Receiver<usize>,
-    operations: HashMap<usize, async_channel::Sender<Value>>,
+    dropped_ids: async_channel::Receiver<SubscriptionId>,
+    operations: HashMap<SubscriptionId, async_channel::Sender<Value>>,
     keep_alive: KeepAliveSettings,
     keep_alive_actor: stream::Boxed<ConnectionCommand>,
 }
@@ -37,7 +37,7 @@ impl ConnectionActor {
     pub(super) fn new(
         connection: Box<dyn ObjectSafeConnection>,
         client: async_channel::Receiver<ConnectionCommand>,
-        dropped_ids: async_channel::Receiver<usize>,
+        dropped_ids: async_channel::Receiver<SubscriptionId>,
         keep_alive: KeepAliveSettings,
     ) -> Self {
         ConnectionActor {
@@ -122,7 +122,7 @@ impl ConnectionActor {
 
         match event {
             event @ (Event::Next { .. } | Event::Error { .. }) => {
-                let Some(id) = event.id().unwrap().parse::<usize>().ok() else {
+                let Some(id) = event.id().and_then(SubscriptionId::from_str) else {
                     return Some(Message::close(Reason::UnknownSubscription));
                 };
 
@@ -142,7 +142,7 @@ impl ConnectionActor {
                 None
             }
             Event::Complete { id } => {
-                let Some(id) = id.parse::<usize>().ok() else {
+                let Some(id) = SubscriptionId::from_str(&id) else {
                     return Some(Message::close(Reason::UnknownSubscription));
                 };
 

src/client/conection_id.rs

@@ -0,0 +1,23 @@
+use std::num::NonZero;
+
+#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)]
+/// An opaque identifier for a subscription
+///
+/// Currently this wraps a `NonZero<usize>` though that may be subject to change
+/// in the future - as a result the underlying type is not exposed publically
+pub struct SubscriptionId(NonZero<usize>);
+
+impl SubscriptionId {
+    pub(super) fn new(id: usize) -> Option<Self> {
+        Some(SubscriptionId(NonZero::new(id)?))
+    }
+
+    #[expect(clippy::inherent_to_string)] // Don't want this to be public, which implementing Display would make it.
+    pub(super) fn to_string(self) -> String {
+        self.0.to_string()
+    }
+
+    pub(super) fn from_str(s: &str) -> Option<Self> {
+        SubscriptionId::new(s.parse::<usize>().ok()?)
+    }
+}

src/client/connection.rs

@@ -1,7 +1,7 @@
 use std::future::Future;
 use std::pin::Pin;
 
-use crate::Error;
+use crate::{Error, SubscriptionId};
 
 /// Abstraction around a websocket connection.
 ///
@@ -63,7 +63,7 @@ impl Message {
         Self::Text(serde_json::to_string(&crate::protocol::Message::Ping::<()>).unwrap())
     }
 
-    pub(crate) fn complete(id: usize) -> Self {
+    pub(crate) fn complete(id: SubscriptionId) -> Self {
         Self::Text(
             serde_json::to_string(&crate::protocol::Message::Complete::<()> { id: id.to_string() })
                 .unwrap(),

src/client/mod.rs

@@ -17,6 +17,7 @@ use crate::{
 
 mod actor;
 mod builder;
+mod conection_id;
 mod connection;
 mod keepalive;
 mod production_future;
@@ -25,6 +26,7 @@ mod stream;
 pub use self::{
     actor::ConnectionActor,
     builder::ClientBuilder,
+    conection_id::SubscriptionId,
     connection::{Connection, Message},
     stream::Subscription,
 };
@@ -55,22 +57,22 @@ pub use self::{
 #[derive(Clone)]
 pub struct Client {
     actor: async_channel::Sender<ConnectionCommand>,
-    drop_sender: async_channel::Sender<usize>,
+    drop_sender: async_channel::Sender<SubscriptionId>,
     subscription_buffer_size: usize,
     next_id: Arc<AtomicUsize>,
 }
 
 impl Client {
     pub(super) fn new_internal(
         actor: async_channel::Sender<ConnectionCommand>,
-        drop_sender: async_channel::Sender<usize>,
+        drop_sender: async_channel::Sender<SubscriptionId>,
         subscription_buffer_size: usize,
     ) -> Self {
         Client {
             actor,
             drop_sender,
             subscription_buffer_size,
-            next_id: Arc::new(AtomicUsize::new(0)),
+            next_id: Arc::new(AtomicUsize::new(1)),
         }
     }
 
@@ -96,6 +98,8 @@ impl Client {
         let request = serde_json::to_string(&message)
             .map_err(|error| Error::Serializing(error.to_string()))?;
 
+        let id = SubscriptionId::new(id).ok_or(Error::ConnectionIdsExhausted)?;
+
         let actor = self.actor.clone();
         actor
             .send(ConnectionCommand::Subscribe {
@@ -134,10 +138,10 @@ pub(super) enum ConnectionCommand {
         /// The full subscribe request as a JSON encoded string.
         request: String,
         sender: async_channel::Sender<Value>,
-        id: usize,
+        id: SubscriptionId,
     },
     Ping,
-    Cancel(usize),
+    Cancel(SubscriptionId),
     Close(u16, String),
 }
 

src/client/stream.rs

@@ -5,7 +5,9 @@ use std::{
 
 use futures_lite::{future, stream, Stream, StreamExt};
 
-use crate::{client::production_future::read_from_producer, graphql::GraphqlOperation, Error};
+use crate::{
+    client::production_future::read_from_producer, graphql::GraphqlOperation, Error, SubscriptionId,
+};
 
 use super::ConnectionCommand;
 
@@ -17,10 +19,10 @@ pub struct Subscription<Operation>
 where
     Operation: GraphqlOperation,
 {
-    pub(in crate::client) id: usize,
+    pub(in crate::client) id: SubscriptionId,
     pub(in crate::client) stream: Option<stream::Boxed<Result<Operation::Response, Error>>>,
     pub(in crate::client) actor: async_channel::Sender<ConnectionCommand>,
-    pub(in crate::client) drop_sender: async_channel::Sender<usize>,
+    pub(in crate::client) drop_sender: async_channel::Sender<SubscriptionId>,
 }
 
 #[pin_project::pinned_drop]

src/error.rs

@@ -25,4 +25,10 @@ pub enum Error {
     /// Sender shutdown error
     #[error("sender shutdown error, reason: {0}")]
     SenderShutdown(String),
+    /// Too many existing connections have been created.
+    ///
+    /// Note that this would require a usize to be exhausted so is quite
+    /// unlikely
+    #[error("connection ID space exhausted.  please restart the client")]
+    ConnectionIdsExhausted,
 }