rpc: Expose the subscription ID for RpcClientT

examples/examples/custom_rpc_client.rs

@@ -16,6 +16,7 @@ use subxt::{
         RpcClientT,
         RpcFuture,
         RpcSubscription,
+        RpcSubscriptionId,
     },
     OnlineClient,
     PolkadotConfig,
@@ -54,7 +55,7 @@ impl RpcClientT for MyLoggingClient {
         sub: &'a str,
         params: Option<Box<RawValue>>,
         unsub: &'a str,
-    ) -> RpcFuture<'a, RpcSubscription> {
+    ) -> RpcFuture<'a, (RpcSubscription, Option<RpcSubscriptionId>)> {
         writeln!(
             self.log.lock().unwrap(),
             "{sub}({}) (unsub: {unsub})",
@@ -68,7 +69,8 @@ impl RpcClientT for MyLoggingClient {
         let res = RawValue::from_string("[]".to_string()).unwrap();
         let stream = futures::stream::once(async move { Ok(res) });
         let stream: Pin<Box<dyn futures::Stream<Item = _> + Send>> = Box::pin(stream);
-        Box::pin(std::future::ready(Ok(stream)))
+        // This subscription does not provide an ID.
+        Box::pin(std::future::ready(Ok((stream, None))))
     }
 }
 

subxt/src/rpc/jsonrpsee_impl.rs

@@ -6,20 +6,25 @@ use super::{
     RpcClientT,
     RpcFuture,
     RpcSubscription,
+    RpcSubscriptionId,
 };
 use crate::error::RpcError;
 use futures::stream::{
     StreamExt,
     TryStreamExt,
 };
-use jsonrpsee::core::{
-    client::{
-        Client,
-        ClientT,
-        SubscriptionClientT,
+use jsonrpsee::{
+    core::{
+        client::{
+            Client,
+            ClientT,
+            SubscriptionClientT,
+            SubscriptionKind,
+        },
+        traits::ToRpcParams,
+        Error as JsonRpseeError,
     },
-    traits::ToRpcParams,
-    Error as JsonRpseeError,
+    types::SubscriptionId,
 };
 use serde_json::value::RawValue;
 
@@ -50,7 +55,7 @@ impl RpcClientT for Client {
         sub: &'a str,
         params: Option<Box<RawValue>>,
         unsub: &'a str,
-    ) -> RpcFuture<'a, RpcSubscription> {
+    ) -> RpcFuture<'a, (RpcSubscription, Option<RpcSubscriptionId>)> {
         Box::pin(async move {
             let sub = SubscriptionClientT::subscribe::<Box<RawValue>, _>(
                 self,
@@ -59,10 +64,17 @@ impl RpcClientT for Client {
                 unsub,
             )
             .await
-            .map_err(|e| RpcError::ClientError(Box::new(e)))?
-            .map_err(|e| RpcError::ClientError(Box::new(e)))
-            .boxed();
-            Ok(sub)
+            .map_err(|e| RpcError::ClientError(Box::new(e)))?;
+
+            let sub_id = match sub.kind() {
+                SubscriptionKind::Subscription(SubscriptionId::Str(id)) => {
+                    Some(id.clone().into_owned())
+                }
+                _ => None,
+            };
+
+            let sub = sub.map_err(|e| RpcError::ClientError(Box::new(e))).boxed();
+            Ok((sub, sub_id))
         })
     }
 }

subxt/src/rpc/mod.rs

@@ -67,6 +67,7 @@ pub use rpc_client_t::{
     RpcClientT,
     RpcFuture,
     RpcSubscription,
+    RpcSubscriptionId,
 };
 
 pub use rpc_client::{

subxt/src/rpc/rpc_client.rs

@@ -5,6 +5,7 @@
 use super::{
     RpcClientT,
     RpcSubscription,
+    RpcSubscriptionId,
 };
 use crate::error::Error;
 use futures::{
@@ -60,8 +61,8 @@ impl RpcClient {
         params: RpcParams,
         unsub: &str,
     ) -> Result<Subscription<Res>, Error> {
-        let sub = self.0.subscribe_raw(sub, params.build(), unsub).await?;
-        Ok(Subscription::new(sub))
+        let (sub, sub_id) = self.0.subscribe_raw(sub, params.build(), unsub).await?;
+        Ok(Subscription::new(sub, sub_id))
     }
 }
 
@@ -166,6 +167,7 @@ impl RpcParams {
 /// [`StreamExt`] extension trait.
 pub struct Subscription<Res> {
     inner: RpcSubscription,
+    sub_id: Option<RpcSubscriptionId>,
     _marker: std::marker::PhantomData<Res>,
 }
 
@@ -179,12 +181,18 @@ impl<Res> std::fmt::Debug for Subscription<Res> {
 }
 
 impl<Res> Subscription<Res> {
-    fn new(inner: RpcSubscription) -> Self {
+    fn new(inner: RpcSubscription, sub_id: Option<RpcSubscriptionId>) -> Self {
         Self {
             inner,
+            sub_id,
             _marker: std::marker::PhantomData,
         }
     }
+
+    /// Obtain the ID associated with this subscription.
+    pub fn subscription_id(&self) -> Option<&RpcSubscriptionId> {
+        self.sub_id.as_ref()
+    }
 }
 
 impl<Res: DeserializeOwned> Subscription<Res> {

subxt/src/rpc/rpc_client_t.rs

@@ -52,7 +52,7 @@ pub trait RpcClientT: Send + Sync + 'static {
         sub: &'a str,
         params: Option<Box<RawValue>>,
         unsub: &'a str,
-    ) -> RpcFuture<'a, RpcSubscription>;
+    ) -> RpcFuture<'a, (RpcSubscription, Option<RpcSubscriptionId>)>;
 }
 
 /// A boxed future that is returned from the [`RpcClientT`] methods.
@@ -62,3 +62,6 @@ pub type RpcFuture<'a, T> =
 /// The inner subscription stream returned from our [`RpcClientT`]'s `subscription` method.
 pub type RpcSubscription =
     Pin<Box<dyn Stream<Item = Result<Box<RawValue>, RpcError>> + Send + 'static>>;
+
+/// The ID associated with the [`RpcClientT`]'s `subscription`.
+pub type RpcSubscriptionId = String;