Add support for wallet remote pubsub consumer

crates/cashu/src/nuts/nut17/mod.rs

@@ -6,7 +6,7 @@ use super::PublicKey;
 use crate::nuts::{
     CurrencyUnit, MeltQuoteBolt11Response, MintQuoteBolt11Response, PaymentMethod, ProofState,
 };
-use crate::quote_id::{QuoteId, QuoteIdError};
+use crate::quote_id::QuoteIdError;
 use crate::MintQuoteBolt12Response;
 
 pub mod ws;
@@ -107,7 +107,10 @@ pub enum WsCommand {
     ProofState,
 }
 
-impl<T> From<MintQuoteBolt12Response<T>> for NotificationPayload<T> {
+impl<T> From<MintQuoteBolt12Response<T>> for NotificationPayload<T>
+where
+    T: Clone,
+{
     fn from(mint_quote: MintQuoteBolt12Response<T>) -> NotificationPayload<T> {
         NotificationPayload::MintQuoteBolt12Response(mint_quote)
     }
@@ -117,7 +120,10 @@ impl<T> From<MintQuoteBolt12Response<T>> for NotificationPayload<T> {
 #[serde(bound = "T: Serialize + DeserializeOwned")]
 #[serde(untagged)]
 /// Subscription response
-pub enum NotificationPayload<T> {
+pub enum NotificationPayload<T>
+where
+    T: Clone,
+{
     /// Proof State
     ProofState(ProofState),
     /// Melt Quote Bolt11 Response
@@ -129,18 +135,22 @@ pub enum NotificationPayload<T> {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Deserialize, Hash, Serialize)]
+#[serde(bound = "T: Serialize + DeserializeOwned")]
 /// A parsed notification
-pub enum NotificationId {
+pub enum NotificationId<T>
+where
+    T: Clone,
+{
     /// ProofState id is a Pubkey
     ProofState(PublicKey),
     /// MeltQuote id is an QuoteId
-    MeltQuoteBolt11(QuoteId),
+    MeltQuoteBolt11(T),
     /// MintQuote id is an QuoteId
-    MintQuoteBolt11(QuoteId),
+    MintQuoteBolt11(T),
     /// MintQuote id is an QuoteId
-    MintQuoteBolt12(QuoteId),
+    MintQuoteBolt12(T),
     /// MintQuote id is an QuoteId
-    MeltQuoteBolt12(QuoteId),
+    MeltQuoteBolt12(T),
 }
 
 /// Kind

crates/cashu/src/nuts/nut17/ws.rs

@@ -36,7 +36,10 @@ pub struct WsUnsubscribeResponse<I> {
 /// subscription
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(bound = "T: Serialize + DeserializeOwned, I: Serialize + DeserializeOwned")]
-pub struct NotificationInner<T, I> {
+pub struct NotificationInner<T, I>
+where
+    T: Clone,
+{
     /// The subscription ID
     #[serde(rename = "subId")]
     pub sub_id: I,

crates/cdk-common/src/pub_sub/error.rs

@@ -12,4 +12,12 @@ pub enum Error {
     /// Parsing error
     #[error("Parsing Error {0}")]
     ParsingError(String),
+
+    /// Internal error
+    #[error("Internal")]
+    Internal(Box<dyn std::error::Error + Send + Sync>),
+
+    /// Not supported
+    #[error("Not supported")]
+    NotSupported,
 }

crates/cdk-common/src/pub_sub/index.rs → crates/cdk-common/src/pub_sub/event.rs

@@ -1,4 +1,6 @@
-//! WS Index
+//! Pubsub Event definition
+//!
+//! The Pubsub Event defines the Topic struct and how an event can be converted to Topics.
 
 use std::fmt::Debug;
 use std::hash::Hash;
@@ -7,12 +9,12 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 
 /// Indexable trait
-pub trait Indexable: Clone {
+pub trait Event: Clone {
     /// Generic Index
     ///
     /// It should be serializable/deserializable to be stored in the database layer and it should
     /// also be sorted in a BTree for in-memory matching
-    type Index: Debug
+    type Topic: Debug
         + Clone
         + Eq
         + PartialEq
@@ -25,5 +27,5 @@ pub trait Indexable: Clone {
         + Serialize;
 
     /// To indexes
-    fn to_indexes(&self) -> Vec<Self::Index>;
+    fn get_topics(&self) -> Vec<Self::Topic>;
 }

crates/cdk-common/src/pub_sub/mod.rs

@@ -12,13 +12,13 @@
 //! Events are also generic that should implement the `Indexable` trait.
 
 mod error;
-pub mod index;
+pub mod event;
 mod pubsub;
 pub mod remote_consumer;
 mod subscriber;
 
 pub use self::error::Error;
-pub use self::index::Indexable;
+pub use self::event::Event;
 pub use self::pubsub::{Pubsub, Topic};
 pub use self::subscriber::SubscriptionRequest;
 
@@ -32,7 +32,7 @@ mod test {
 
     use super::pubsub::Topic;
     use super::subscriber::SubscriptionRequest;
-    use super::{Error, Indexable, Pubsub};
+    use super::{Error, Event, Pubsub};
 
     #[derive(Clone, Debug, Serialize, Deserialize)]
     pub struct Message {
@@ -46,10 +46,10 @@ mod test {
         Bar(u64),
     }
 
-    impl Indexable for Message {
-        type Index = IndexTest;
+    impl Event for Message {
+        type Topic = IndexTest;
 
-        fn to_indexes(&self) -> Vec<Self::Index> {
+        fn get_topics(&self) -> Vec<Self::Topic> {
             vec![IndexTest::Foo(self.foo), IndexTest::Bar(self.bar)]
         }
     }
@@ -67,7 +67,7 @@ mod test {
 
         async fn fetch_events(
             &self,
-            indexes: Vec<<Self::Event as Indexable>::Index>,
+            indexes: Vec<<Self::Event as Event>::Topic>,
             sub_name: Self::SubscriptionName,
             reply_to: mpsc::Sender<(Self::SubscriptionName, Self::Event)>,
         ) {
@@ -83,7 +83,7 @@ mod test {
         /// Store events or replace them
         async fn store_events(&self, event: Self::Event) {
             let mut storage = self.storage.write().unwrap();
-            for index in event.to_indexes() {
+            for index in event.get_topics() {
                 storage.insert(index, event.clone());
             }
         }
@@ -96,11 +96,11 @@ mod test {
     }
 
     impl SubscriptionRequest for SubscriptionReq {
-        type Index = IndexTest;
+        type Topic = IndexTest;
 
         type SubscriptionName = String;
 
-        fn try_get_indexes(&self) -> Result<Vec<Self::Index>, Error> {
+        fn try_get_topics(&self) -> Result<Vec<Self::Topic>, Error> {
             Ok(vec![match self {
                 SubscriptionReq::Bar(n) => IndexTest::Bar(*n),
                 SubscriptionReq::Foo(n) => IndexTest::Foo(*n),

crates/cdk-common/src/pub_sub/pubsub.rs

@@ -11,7 +11,7 @@ use serde::de::DeserializeOwned;
 use serde::Serialize;
 use tokio::sync::mpsc;
 
-use super::index::Indexable;
+use super::event::Event;
 use super::subscriber::{ActiveSubscription, SubscriptionRequest};
 use super::Error;
 
@@ -41,12 +41,12 @@ pub trait Topic: Send + Sync {
         + Serialize;
 
     /// An Event should be Indexable
-    type Event: Indexable + Debug + Send + Sync + DeserializeOwned + Serialize;
+    type Event: Event + Debug + Send + Sync + DeserializeOwned + Serialize;
 
     /// Called when a new subscription is created. The function is responsible to not yield the same
     async fn fetch_events(
         &self,
-        indexes: Vec<<Self::Event as Indexable>::Index>,
+        indexes: Vec<<Self::Event as Event>::Topic>,
         sub_name: Self::SubscriptionName,
         reply_to: mpsc::Sender<(Self::SubscriptionName, Self::Event)>,
     );
@@ -63,7 +63,7 @@ pub type IndexTree<P> = Arc<
     RwLock<
         BTreeMap<
             // Index with a subscription unique ID
-            (<<P as Topic>::Event as Indexable>::Index, usize),
+            (<<P as Topic>::Event as Event>::Topic, usize),
             (
                 <P as Topic>::SubscriptionName, // Subscription ID, as given by the client, more like a name
                 mpsc::Sender<(<P as Topic>::SubscriptionName, <P as Topic>::Event)>, // Sender
@@ -111,7 +111,7 @@ where
     ) -> Result<(), Error> {
         let index_storage = listeners_index.read().map_err(|_| Error::Poison)?;
         let mut sent = HashSet::new();
-        for index in event.to_indexes() {
+        for index in event.get_topics() {
             for ((subscription_index, unique_id), (subscription_id, sender)) in
                 index_storage.range((index.clone(), 0)..)
             {
@@ -170,7 +170,7 @@ where
     ) -> Result<ActiveSubscription<P>, Error>
     where
         I: SubscriptionRequest<
-            Index = <P::Event as Indexable>::Index,
+            Topic = <P::Event as Event>::Topic,
             SubscriptionName = P::SubscriptionName,
         >,
     {
@@ -183,7 +183,7 @@ where
             .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
 
         let subscription_name = request.subscription_name();
-        let subscribed_to = request.try_get_indexes()?;
+        let subscribed_to = request.try_get_topics()?;
 
         for index in subscribed_to.iter() {
             index_storage.insert(
@@ -216,7 +216,7 @@ where
     pub fn subscribe<I>(&self, request: I) -> Result<ActiveSubscription<P>, Error>
     where
         I: SubscriptionRequest<
-            Index = <P::Event as Indexable>::Index,
+            Topic = <P::Event as Event>::Topic,
             SubscriptionName = P::SubscriptionName,
         >,
     {

crates/cdk-common/src/pub_sub/remote_consumer.rs

@@ -4,25 +4,26 @@
 use std::collections::HashMap;
 use std::sync::atomic::AtomicBool;
 use std::sync::{Arc, RwLock};
+use std::time::Duration;
 
 use tokio::sync::mpsc;
+use tokio::time::sleep;
 
 use super::pubsub::Topic;
 use super::subscriber::{ActiveSubscription, SubscriptionRequest};
-use super::{Error, Indexable, Pubsub};
-
-type ActiveSubscriptions<T> = RwLock<
-    HashMap<
-        <T as Topic>::SubscriptionName,
-        (
-            Vec<<<T as Topic>::Event as Indexable>::Index>,
-            ActiveSubscription<T>,
-        ),
-    >,
->;
+use super::{Error, Event, Pubsub};
+
+type UniqueSubscriptions<T> =
+    RwLock<HashMap<<<T as Topic>::Event as Event>::Topic, (usize, ActiveSubscription<T>)>>;
+
+type ActiveSubscriptions<T> =
+    RwLock<HashMap<<T as Topic>::SubscriptionName, Vec<<<T as Topic>::Event as Event>::Topic>>>;
 
 type InternalSender<T> = mpsc::Sender<(<T as Topic>::SubscriptionName, <T as Topic>::Event)>;
 
+const LONG_CONNECTION_SLEEP_MS: u64 = 10;
+const POLL_SLEEP_SECS: u64 = 2;
+
 /// Subscription consumer
 pub struct Consumer<T>
 where
@@ -31,24 +32,67 @@ where
     transport: T,
     pubsub_internal_sender: InternalSender<T::Topic>,
     inner_pubsub: Pubsub<T::Topic>,
+    unique_subscriptions: UniqueSubscriptions<T::Topic>,
     subscriptions: ActiveSubscriptions<T::Topic>,
     send_to_transport_loop: RwLock<mpsc::Sender<MessageToTransportLoop<T::Topic>>>,
     still_running: AtomicBool,
 }
 
+/// Remote consumer
+pub struct RemoteActiveConsumer<T>
+where
+    T: Transport + 'static, {}
+
+struct InternalConversion<T>
+where
+    T: Transport + 'static,
+{
+    name: <T::Topic as Topic>::SubscriptionName,
+    index: <<T::Topic as Topic>::Event as Event>::Topic,
+}
+
+impl<T> Clone for InternalConversion<T>
+where
+    T: Transport + 'static,
+{
+    fn clone(&self) -> Self {
+        Self {
+            name: self.name.clone(),
+            index: self.index.clone(),
+        }
+    }
+}
+
+impl<T> SubscriptionRequest for InternalConversion<T>
+where
+    T: Transport + 'static,
+{
+    type Topic = <<T::Topic as Topic>::Event as Event>::Topic;
+
+    type SubscriptionName = <T::Topic as Topic>::SubscriptionName;
+
+    fn subscription_name(&self) -> Self::SubscriptionName {
+        self.name.to_owned()
+    }
+
+    fn try_get_topics(&self) -> Result<Vec<Self::Topic>, Error> {
+        Ok(vec![self.index.clone()])
+    }
+}
+
 impl<T> Consumer<T>
 where
     T: Transport + 'static,
 {
     /// Creates a new instance
-    pub async fn new(transport: T) -> Arc<Self> {
-        let (sender, _) = mpsc::channel(10_000);
+    pub fn new<F>(transport: T, inner_pubsub: Pubsub<T::Topic>) -> Arc<Self> {
         let this = Arc::new(Self {
             transport,
-            inner_pubsub: T::new_pubsub().await,
+            inner_pubsub,
             pubsub_internal_sender: mpsc::channel(10_000).0,
             subscriptions: Default::default(),
-            send_to_transport_loop: RwLock::new(sender),
+            unique_subscriptions: Default::default(),
+            send_to_transport_loop: RwLock::new(mpsc::channel(10_000).0),
             still_running: true.into(),
         });
 
@@ -58,15 +102,72 @@ where
     }
 
     async fn connection_loop(instance: Arc<Self>) {
-        loop {
-            let (sender, receiver) = mpsc::channel(10_000);
+        let mut long_connection_supported = true;
+        let mut poll_supported = true;
 
+        loop {
+            if (!long_connection_supported && !poll_supported)
+                || !instance
+                    .still_running
+                    .load(std::sync::atomic::Ordering::Relaxed)
             {
-                let mut shared_sender = instance.send_to_transport_loop.write().unwrap();
-                *shared_sender = sender;
+                break;
             }
 
-            instance.transport.long_connection(receiver).await;
+            if long_connection_supported {
+                let (sender, receiver) = mpsc::channel(10_000);
+
+                {
+                    let mut shared_sender = instance.send_to_transport_loop.write().unwrap();
+                    *shared_sender = sender;
+                }
+
+                let current_subscriptions = instance
+                    .unique_subscriptions
+                    .read()
+                    .expect("xxx")
+                    .keys()
+                    .map(|x| x.clone())
+                    .collect::<Vec<_>>();
+
+                if let Err(err) = instance
+                    .transport
+                    .long_connection(receiver, current_subscriptions)
+                    .await
+                {
+                    if matches!(err, Error::NotSupported) {
+                        long_connection_supported = false;
+                    }
+                    tracing::error!("Long connection failed with error {:?}", err);
+                }
+                sleep(Duration::from_millis(LONG_CONNECTION_SLEEP_MS)).await;
+            }
+
+            if poll_supported {
+                let current_subscriptions = instance
+                    .unique_subscriptions
+                    .read()
+                    .expect("xxx")
+                    .iter()
+                    .map(|(key, value)| (value.1.name().clone(), key.clone()))
+                    .collect::<Vec<_>>();
+
+                if let Err(err) = instance
+                    .transport
+                    .poll(
+                        current_subscriptions,
+                        instance.pubsub_internal_sender.clone(),
+                    )
+                    .await
+                {
+                    if matches!(err, Error::NotSupported) {
+                        poll_supported = false;
+                    }
+                    tracing::error!("Polling failed with error {:?}", err);
+                }
+
+                sleep(Duration::from_secs(POLL_SLEEP_SECS)).await;
+            }
         }
     }
 
@@ -82,44 +183,61 @@ where
     ///    broadcasat the event.
     ///
     ///
-    pub fn subscribe<I>(&self, request: I) -> Result<(), Error>
+    pub fn subscribe<I>(&self, request: I) -> Result<RemoteActiveConsumer<T>, Error>
     where
         I: SubscriptionRequest<
-            Index = <<T::Topic as Topic>::Event as Indexable>::Index,
+            Topic = <<T::Topic as Topic>::Event as Event>::Topic,
             SubscriptionName = <T::Topic as Topic>::SubscriptionName,
         >,
     {
-        let transport_loop = self
-            .send_to_transport_loop
-            .read()
+        let subscription_name = request.subscription_name();
+        let indexes = request.try_get_topics()?;
+
+        let mut unique_subscriptions = self
+            .unique_subscriptions
+            .write()
             .map_err(|_| Error::Poison)?;
         let mut subscriptions = self.subscriptions.write().map_err(|_| Error::Poison)?;
-        let subscription_name = request.subscription_name();
-        let indexes = request.try_get_indexes()?;
 
         if subscriptions.get(&subscription_name).is_some() {
             return Err(Error::AlreadySubscribed);
         }
 
-        subscriptions.insert(
-            subscription_name.clone(),
-            (
-                indexes.clone(),
-                self.inner_pubsub.subscribe_with(
-                    request,
-                    self.pubsub_internal_sender.clone(),
-                    None,
-                )?,
-            ),
-        );
-        drop(subscriptions);
+        for index in indexes.iter() {
+            if let Some(subscription) = unique_subscriptions.get_mut(&index) {
+                subscription.0 = subscription.0 + 1;
+            } else {
+                unique_subscriptions.insert(
+                    index.clone(),
+                    (
+                        0,
+                        self.inner_pubsub.subscribe_with(
+                            InternalConversion::<T> {
+                                name: self.transport.new_name(),
+                                index: index.clone(),
+                            },
+                            self.pubsub_internal_sender.clone(),
+                            None,
+                        )?,
+                    ),
+                );
 
-        let _ = transport_loop.try_send(MessageToTransportLoop::Subscribe((
-            subscription_name,
-            indexes,
-        )));
+                // new subscription is created, so the connection worker should be notified
+                let _ = self
+                    .send_to_transport_loop
+                    .read()
+                    .map_err(|_| Error::Poison)?
+                    .try_send(MessageToTransportLoop::Subscribe((
+                        subscription_name.clone(),
+                        indexes.clone(),
+                    )));
+            }
+        }
 
-        Ok(())
+        subscriptions.insert(subscription_name, indexes);
+        drop(subscriptions);
+
+        todo!()
     }
 }
 
@@ -141,7 +259,7 @@ where
     T: Topic + 'static,
 {
     /// Add a subscription
-    Subscribe((T::SubscriptionName, Vec<<T::Event as Indexable>::Index>)),
+    Subscribe((T::SubscriptionName, Vec<<T::Event as Event>::Topic>)),
     /// Desuscribe
     Desuscribe(T::SubscriptionName),
     /// Exit the loop
@@ -154,19 +272,28 @@ pub trait Transport: Send + Sync {
     /// Topic
     type Topic: Topic + Clone + Sync + Send;
 
-    /// Creates a new pubsub topic producer
-    async fn new_pubsub() -> Pubsub<Self::Topic>;
+    /// Create a new subscription name
+    fn new_name(&self) -> <Self::Topic as Topic>::SubscriptionName;
 
     /// Open a long connection
     async fn long_connection(
         &self,
         subscribe_changes: mpsc::Receiver<MessageToTransportLoop<Self::Topic>>,
-    ) where
+        topics: Vec<<<Self::Topic as Topic>::Event as Event>::Topic>,
+    ) -> Result<(), Error>
+    where
         Self: Sized;
 
     /// Poll on demand
     async fn poll(
         &self,
-        index: Vec<<<Self::Topic as Topic>::Event as Indexable>::Index>,
-    ) -> Result<Vec<Self::Topic>, Error>;
+        topics: Vec<(
+            <Self::Topic as Topic>::SubscriptionName,
+            <<Self::Topic as Topic>::Event as Event>::Topic,
+        )>,
+        reply_to: mpsc::Sender<(
+            <Self::Topic as Topic>::SubscriptionName,
+            <Self::Topic as Topic>::Event,
+        )>,
+    ) -> Result<(), Error>;
 }

crates/cdk-common/src/pub_sub/subscriber.rs

@@ -6,18 +6,18 @@ use tokio::sync::mpsc;
 
 use super::pubsub::{IndexTree, Topic};
 use super::Error;
-use crate::pub_sub::index::Indexable;
+use crate::pub_sub::event::Event;
 
 /// Subscription request
 pub trait SubscriptionRequest: Clone {
     /// Indexes
-    type Index;
+    type Topic;
 
     /// Subscription name
     type SubscriptionName;
 
     /// Try to get indexes from the request
-    fn try_get_indexes(&self) -> Result<Vec<Self::Index>, Error>;
+    fn try_get_topics(&self) -> Result<Vec<Self::Topic>, Error>;
 
     /// Get the subscription name
     fn subscription_name(&self) -> Self::SubscriptionName;
@@ -32,7 +32,7 @@ where
     name: P::SubscriptionName,
     active_subscribers: Arc<AtomicUsize>,
     indexes: IndexTree<P>,
-    subscribed_to: Vec<<P::Event as Indexable>::Index>,
+    subscribed_to: Vec<<P::Event as Event>::Topic>,
     receiver: Option<mpsc::Receiver<(P::SubscriptionName, P::Event)>>,
 }
 
@@ -46,7 +46,7 @@ where
         name: P::SubscriptionName,
         active_subscribers: Arc<AtomicUsize>,
         indexes: IndexTree<P>,
-        subscribed_to: Vec<<P::Event as Indexable>::Index>,
+        subscribed_to: Vec<<P::Event as Event>::Topic>,
         receiver: Option<mpsc::Receiver<(P::SubscriptionName, P::Event)>>,
     ) -> Self {
         Self {

crates/cdk-common/src/subscription.rs

@@ -61,7 +61,7 @@ impl Deref for SubId {
 }
 
 impl SubscriptionRequest for IndexableParams {
-    type Index = NotificationId;
+    type Topic = NotificationId<QuoteId>;
 
     type SubscriptionName = SubId;
 
@@ -69,7 +69,7 @@ impl SubscriptionRequest for IndexableParams {
         self.0.id.clone()
     }
 
-    fn try_get_indexes(&self) -> Result<Vec<Self::Index>, Error> {
+    fn try_get_topics(&self) -> Result<Vec<Self::Topic>, Error> {
         self.0
             .filters
             .iter()

crates/cdk-ffi/src/wallet.rs

@@ -351,7 +351,7 @@ impl Wallet {
     ) -> Result<std::sync::Arc<ActiveSubscription>, FfiError> {
         let cdk_params: cdk::nuts::nut17::Params<SubId> = params.clone().into();
         let sub_id = cdk_params.id.to_string();
-        let active_sub = self.inner.subscribe(cdk_params).await;
+        let active_sub = self.inner.subscribe(cdk_params);
         Ok(std::sync::Arc::new(ActiveSubscription::new(
             active_sub, sub_id,
         )))

crates/cdk/src/event.rs

@@ -0,0 +1,111 @@
+//! Mint event types
+use std::fmt::Debug;
+use std::hash::Hash;
+
+use cdk_common::nut17::NotificationId;
+use cdk_common::pub_sub::Event;
+use cdk_common::{
+    MeltQuoteBolt11Response, MintQuoteBolt11Response, MintQuoteBolt12Response, NotificationPayload,
+    ProofState,
+};
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
+
+/// Simple wrapper over NotificationPayload<QuoteId> which is a foreign type
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(bound = "T: Serialize + DeserializeOwned")]
+pub struct MintEvent<T>(NotificationPayload<T>)
+where
+    T: Clone;
+
+impl<T> From<ProofState> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn from(value: ProofState) -> Self {
+        Self(NotificationPayload::ProofState(value))
+    }
+}
+
+#[allow(clippy::from_over_into)]
+impl<T> Into<NotificationPayload<T>> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn into(self) -> NotificationPayload<T> {
+        self.0
+    }
+}
+
+impl<T> MintEvent<T>
+where
+    T: Clone,
+{
+    /// New instance
+    pub fn new(t: NotificationPayload<T>) -> Self {
+        Self(t)
+    }
+
+    /// Get inner
+    pub fn inner(self) -> NotificationPayload<T> {
+        self.0
+    }
+}
+
+impl<T> From<NotificationPayload<T>> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn from(value: NotificationPayload<T>) -> Self {
+        Self(value)
+    }
+}
+
+impl<T> From<MintQuoteBolt11Response<T>> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn from(value: MintQuoteBolt11Response<T>) -> Self {
+        Self(NotificationPayload::MintQuoteBolt11Response(value))
+    }
+}
+
+impl<T> From<MeltQuoteBolt11Response<T>> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn from(value: MeltQuoteBolt11Response<T>) -> Self {
+        Self(NotificationPayload::MeltQuoteBolt11Response(value))
+    }
+}
+
+impl<T> From<MintQuoteBolt12Response<T>> for MintEvent<T>
+where
+    T: Clone,
+{
+    fn from(value: MintQuoteBolt12Response<T>) -> Self {
+        Self(NotificationPayload::MintQuoteBolt12Response(value))
+    }
+}
+
+impl<T> Event for MintEvent<T>
+where
+    T: Clone + Serialize + DeserializeOwned + Debug + Ord + Hash + Send + Sync,
+{
+    type Topic = NotificationId<T>;
+
+    fn get_topics(&self) -> Vec<Self::Topic> {
+        vec![match &self.0 {
+            NotificationPayload::MeltQuoteBolt11Response(r) => {
+                NotificationId::MeltQuoteBolt11(r.quote.to_owned())
+            }
+            NotificationPayload::MintQuoteBolt11Response(r) => {
+                NotificationId::MintQuoteBolt11(r.quote.to_owned())
+            }
+            NotificationPayload::MintQuoteBolt12Response(r) => {
+                NotificationId::MintQuoteBolt12(r.quote.to_owned())
+            }
+            NotificationPayload::ProofState(p) => NotificationId::ProofState(p.y.to_owned()),
+        }]
+    }
+}

crates/cdk/src/lib.rs

@@ -41,6 +41,7 @@ pub use cdk_common::{
 #[cfg(all(any(feature = "wallet", feature = "mint"), feature = "auth"))]
 pub use oidc_client::OidcClient;
 
+mod event;
 pub mod fees;
 
 #[doc(hidden)]

crates/cdk/src/mint/mod.rs

@@ -20,7 +20,7 @@ use cdk_signatory::signatory::{Signatory, SignatoryKeySet};
 use futures::StreamExt;
 #[cfg(feature = "auth")]
 use nut21::ProtectedEndpoint;
-use pubsub_manager::PubSubManager;
+use subscription::PubSubManager;
 use tokio::sync::{Mutex, Notify};
 use tokio::task::{JoinHandle, JoinSet};
 use tracing::instrument;
@@ -41,8 +41,8 @@ mod keysets;
 mod ln;
 mod melt;
 mod proof_writer;
-mod pubsub_manager;
 mod start_up_check;
+mod subscription;
 mod swap;
 mod verification;
 

crates/cdk/src/mint/proof_writer.rs

@@ -5,7 +5,7 @@ use std::sync::Arc;
 use cdk_common::database::{self, DynMintDatabase, MintTransaction};
 use cdk_common::{Error, Proofs, ProofsMethods, PublicKey, QuoteId, State};
 
-use super::pubsub_manager::PubSubManager;
+use super::subscription::PubSubManager;
 
 type Tx<'a, 'b> = Box<dyn MintTransaction<'a, database::Error> + Send + Sync + 'b>;
 

crates/cdk/src/mint/pubsub_manager.rs → crates/cdk/src/mint/subscription.rs

@@ -6,83 +6,17 @@ use std::sync::Arc;
 use cdk_common::database::DynMintDatabase;
 use cdk_common::mint::MintQuote;
 use cdk_common::nut17::NotificationId;
-use cdk_common::pub_sub::{Indexable, Pubsub, Topic};
+use cdk_common::pub_sub::{Event, Pubsub, Topic};
 use cdk_common::subscription::SubId;
 use cdk_common::{
     Amount, BlindSignature, MeltQuoteBolt11Response, MeltQuoteState, MintQuoteBolt11Response,
-    MintQuoteBolt12Response, MintQuoteState, NotificationPayload, PaymentMethod, ProofState,
-    PublicKey, QuoteId,
+    MintQuoteBolt12Response, MintQuoteState, PaymentMethod, ProofState, PublicKey, QuoteId,
 };
-use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc;
 
-/// Simple wrapper over NotificationPayload<QuoteId> which is a foreign type
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct MintEvent(NotificationPayload<QuoteId>);
-
-#[allow(clippy::from_over_into)]
-impl Into<NotificationPayload<QuoteId>> for MintEvent {
-    fn into(self) -> NotificationPayload<QuoteId> {
-        self.0
-    }
-}
-
-impl MintEvent {
-    /// Get inner
-    pub fn inner(self) -> NotificationPayload<QuoteId> {
-        self.0
-    }
-}
-
-impl From<NotificationPayload<QuoteId>> for MintEvent {
-    fn from(value: NotificationPayload<QuoteId>) -> Self {
-        Self(value)
-    }
-}
-
-impl From<ProofState> for MintEvent {
-    fn from(value: ProofState) -> Self {
-        Self(NotificationPayload::ProofState(value))
-    }
-}
-
-impl From<MintQuoteBolt11Response<QuoteId>> for MintEvent {
-    fn from(value: MintQuoteBolt11Response<QuoteId>) -> Self {
-        Self(NotificationPayload::MintQuoteBolt11Response(value))
-    }
-}
-
-impl From<MeltQuoteBolt11Response<QuoteId>> for MintEvent {
-    fn from(value: MeltQuoteBolt11Response<QuoteId>) -> Self {
-        Self(NotificationPayload::MeltQuoteBolt11Response(value))
-    }
-}
-
-impl From<MintQuoteBolt12Response<QuoteId>> for MintEvent {
-    fn from(value: MintQuoteBolt12Response<QuoteId>) -> Self {
-        Self(NotificationPayload::MintQuoteBolt12Response(value))
-    }
-}
-
-impl Indexable for MintEvent {
-    type Index = NotificationId;
-
-    fn to_indexes(&self) -> Vec<Self::Index> {
-        vec![match &self.0 {
-            NotificationPayload::MeltQuoteBolt11Response(r) => {
-                NotificationId::MeltQuoteBolt11(r.quote.to_owned())
-            }
-            NotificationPayload::MintQuoteBolt11Response(r) => {
-                NotificationId::MintQuoteBolt11(r.quote.to_owned())
-            }
-            NotificationPayload::MintQuoteBolt12Response(r) => {
-                NotificationId::MintQuoteBolt12(r.quote.to_owned())
-            }
-            NotificationPayload::ProofState(p) => NotificationId::ProofState(p.y.to_owned()),
-        }]
-    }
-}
+use crate::event::MintEvent;
 
+// Mint subtopics
 pub struct MintSubTopics {
     db: DynMintDatabase,
 }
@@ -90,8 +24,8 @@ pub struct MintSubTopics {
 impl MintSubTopics {
     async fn get_events_from_db_legacy(
         &self,
-        request: &[NotificationId],
-    ) -> Result<Vec<MintEvent>, String> {
+        request: &[NotificationId<QuoteId>],
+    ) -> Result<Vec<MintEvent<QuoteId>>, String> {
         let mut to_return = vec![];
         let mut public_keys: Vec<PublicKey> = Vec::new();
         let mut melt_queries = Vec::new();
@@ -181,11 +115,11 @@ impl MintSubTopics {
 impl Topic for MintSubTopics {
     type SubscriptionName = SubId;
 
-    type Event = MintEvent;
+    type Event = MintEvent<QuoteId>;
 
     async fn fetch_events(
         &self,
-        indexes: Vec<<Self::Event as Indexable>::Index>,
+        indexes: Vec<<Self::Event as Event>::Topic>,
         sub_name: Self::SubscriptionName,
         reply_to: mpsc::Sender<(Self::SubscriptionName, Self::Event)>,
     ) {

crates/cdk/src/wallet/mod.rs

@@ -192,8 +192,7 @@ impl Wallet {
     /// Subscribe to events
     pub async fn subscribe<T: Into<Params>>(&self, query: T) -> ActiveSubscription {
         self.subscription
-            .subscribe(self.mint_url.clone(), query.into(), Arc::new(self.clone()))
-            .await
+            .subscribe(self.mint_url.clone(), query.into())
     }
 
     /// Fee required for proof set

crates/cdk/src/wallet/subscription.rs

@@ -0,0 +1,303 @@
+//! Client for subscriptions
+//!
+//! Mint servers can send notifications to clients about changes in the state,
+//! according to NUT-17, using the WebSocket protocol. This module provides a
+//! subscription manager that allows clients to subscribe to notifications from
+//! multiple mint servers using WebSocket or with a poll-based system, using
+//! the HTTP client.
+use std::collections::HashMap;
+use std::fmt::Debug;
+use std::sync::{Arc, RwLock};
+
+use cdk_common::nut17::NotificationId;
+use cdk_common::pub_sub::remote_consumer::{Consumer, MessageToTransportLoop, Transport};
+use cdk_common::pub_sub::Error as PubsubError;
+use cdk_common::pub_sub::Topic;
+use cdk_common::pub_sub::{Event, Pubsub};
+use cdk_common::subscription::{Params, SubId};
+use cdk_common::CheckStateRequest;
+use tokio::sync::mpsc;
+#[cfg(target_arch = "wasm32")]
+use wasm_bindgen_futures;
+
+use crate::event::MintEvent;
+use crate::mint_url::MintUrl;
+use crate::wallet::MintConnector;
+
+type WsSubscriptionBody = (mpsc::Sender<NotificationPayload>, Params);
+
+/// Subscription manager
+///
+/// This structure should be instantiated once per wallet at most. It is
+/// cloneable since all its members are Arcs.
+///
+/// The main goal is to provide a single interface to manage multiple
+/// subscriptions to many servers to subscribe to events. If supported, the
+/// WebSocket method is used to subscribe to server-side events. Otherwise, a
+/// poll-based system is used, where a background task fetches information about
+/// the resource every few seconds and notifies subscribers of any change
+/// upstream.
+///
+/// The subscribers have a simple-to-use interface, receiving an
+/// ActiveSubscription struct, which can be used to receive updates and to
+/// unsubscribe from updates automatically on the drop.
+#[derive(Clone)]
+pub struct SubscriptionManager {
+    all_connections: Arc<RwLock<HashMap<MintUrl, Arc<Consumer<SubscriptionClient>>>>>,
+    http_client: Arc<dyn MintConnector + Send + Sync>,
+    prefer_http: bool,
+}
+
+impl Debug for SubscriptionManager {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Subscription Manager connected to {:?}",
+            self.all_connections
+                .write()
+                .map(|connections| connections.keys().cloned().collect::<Vec<_>>())
+                .unwrap_or_default()
+        )
+    }
+}
+
+impl SubscriptionManager {
+    /// Create a new subscription manager
+    pub fn new(http_client: Arc<dyn MintConnector + Send + Sync>, prefer_http: bool) -> Self {
+        Self {
+            all_connections: Arc::new(RwLock::new(HashMap::new())),
+            http_client,
+            prefer_http,
+        }
+    }
+
+    /// Subscribe to updates from a mint server with a given filter
+    pub fn subscribe(&self, mint_url: MintUrl, filter: Params) -> ActiveSubscription {
+        let y = self
+            .all_connections
+            .write()
+            .expect("xxx")
+            .entry(mint_url.clone())
+            .or_insert_with(|| {
+                Arc::new(Consumer::new(
+                    SubscriptionClient {
+                        mint_url,
+                        http_client: self.http_client.clone(),
+                    },
+                    || SubId("test".to_owned()),
+                ))
+            })
+            .clone();
+
+        let y = y.subscribe(filter)?;
+
+        todo!()
+    }
+}
+
+/// MintSubTopics
+#[derive(Clone)]
+pub struct MintSubTopics {}
+
+#[async_trait::async_trait]
+impl Topic for MintSubTopics {
+    type SubscriptionName = SubId;
+
+    type Event = MintEvent<String>;
+
+    async fn fetch_events(
+        &self,
+        _indexes: Vec<<Self::Event as Event>::Topic>,
+        _sub_name: Self::SubscriptionName,
+        _reply_to: mpsc::Sender<(Self::SubscriptionName, Self::Event)>,
+    ) {
+    }
+
+    /// Store events or replace them
+    async fn store_events(&self, _event: Self::Event) {}
+}
+
+/// Subscription client
+///
+/// If the server supports WebSocket subscriptions, this client will be used,
+/// otherwise the HTTP pool and pause will be used (which is the less efficient
+/// method).
+#[derive(Debug)]
+pub struct SubscriptionClient {
+    http_client: Arc<dyn MintConnector + Send + Sync>,
+    mint_url: MintUrl,
+}
+
+#[async_trait::async_trait]
+impl Transport for SubscriptionClient {
+    type Topic = MintSubTopics;
+
+    async fn new_pubsub() -> Pubsub<Self::Topic> {
+        todo!()
+    }
+
+    async fn long_connection(
+        &self,
+        _subscribe_changes: mpsc::Receiver<MessageToTransportLoop<Self::Topic>>,
+        _topics: Vec<<<Self::Topic as Topic>::Event as Event>::Topic>,
+    ) -> Result<(), PubsubError>
+    where
+        Self: Sized,
+    {
+        Err(cdk_common::pub_sub::Error::NotSupported)
+    }
+
+    /// Poll on demand
+    async fn poll(
+        &self,
+        topics: Vec<(
+            <Self::Topic as Topic>::SubscriptionName,
+            <<Self::Topic as Topic>::Event as Event>::Topic,
+        )>,
+        reply_to: mpsc::Sender<(
+            <Self::Topic as Topic>::SubscriptionName,
+            <Self::Topic as Topic>::Event,
+        )>,
+    ) -> Result<(), PubsubError> {
+        let proofs = topics
+            .iter()
+            .filter_map(|x| match &x.1 {
+                NotificationId::ProofState(p) => Some((*p, x.0.clone())),
+                _ => None,
+            })
+            .collect::<HashMap<_, _>>();
+
+        if !proofs.is_empty() {
+            for state in self
+                .http_client
+                .post_check_state(CheckStateRequest {
+                    ys: proofs.keys().map(|x| x.clone()).collect::<Vec<_>>(),
+                })
+                .await
+                .map_err(|e| PubsubError::Internal(Box::new(e)))?
+                .states
+            {
+                if let Some(name) = proofs.get(&state.y) {
+                    let _ = reply_to.try_send((
+                        name.clone(),
+                        MintEvent::new(NotificationPayload::ProofState(state)),
+                    ));
+                }
+            }
+        }
+
+        for (name, topic) in topics
+            .into_iter()
+            .filter(|x| !matches!(x.1, NotificationId::ProofState(_)))
+        {
+            match topic {
+                NotificationId::MintQuoteBolt11(id) => {
+                    let response = match self.http_client.get_mint_quote_status(&id).await {
+                        Ok(success) => success,
+                        Err(err) => {
+                            tracing::error!("Error with MintBolt11 {} with {:?}", id, err);
+                            continue;
+                        }
+                    };
+
+                    let _ = reply_to.try_send((
+                        name,
+                        MintEvent::new(NotificationPayload::MintQuoteBolt11Response(response)),
+                    ));
+                }
+                NotificationId::MeltQuoteBolt11(id) => {
+                    let response = match self.http_client.get_melt_quote_status(&id).await {
+                        Ok(success) => success,
+                        Err(err) => {
+                            tracing::error!("Error with MeltBolt11 {} with {:?}", id, err);
+                            continue;
+                        }
+                    };
+
+                    let _ = reply_to.try_send((
+                        name,
+                        MintEvent::new(NotificationPayload::MeltQuoteBolt11Response(response)),
+                    ));
+                }
+                NotificationId::MintQuoteBolt12(id) => {
+                    let response = match self.http_client.get_mint_quote_bolt12_status(&id).await {
+                        Ok(success) => success,
+                        Err(err) => {
+                            tracing::error!("Error with MintBolt12 {} with {:?}", id, err);
+                            continue;
+                        }
+                    };
+
+                    let _ = reply_to.try_send((
+                        name,
+                        MintEvent::new(NotificationPayload::MintQuoteBolt12Response(response)),
+                    ));
+                }
+                NotificationId::MeltQuoteBolt12(id) => {
+                    let response = match self.http_client.get_melt_bolt12_quote_status(&id).await {
+                        Ok(success) => success,
+                        Err(err) => {
+                            tracing::error!("Error with MeltBolt12 {} with {:?}", id, err);
+                            continue;
+                        }
+                    };
+
+                    let _ = reply_to.try_send((
+                        name,
+                        MintEvent::new(NotificationPayload::MeltQuoteBolt11Response(
+                            response.into(),
+                        )),
+                    ));
+                }
+                _ => {}
+            }
+        }
+
+        Ok(())
+    }
+}
+
+type NotificationPayload = crate::nuts::NotificationPayload<String>;
+
+/// Active Subscription
+pub struct ActiveSubscription {
+    sub_id: Option<SubId>,
+    on_drop_notif: mpsc::Sender<SubId>,
+    receiver: mpsc::Receiver<NotificationPayload>,
+}
+
+impl ActiveSubscription {
+    fn new(
+        receiver: mpsc::Receiver<NotificationPayload>,
+        sub_id: SubId,
+        on_drop_notif: mpsc::Sender<SubId>,
+    ) -> Self {
+        Self {
+            sub_id: Some(sub_id),
+            on_drop_notif,
+            receiver,
+        }
+    }
+
+    /// Try to receive a notification
+    pub fn try_recv(&mut self) -> Result<Option<NotificationPayload>, Error> {
+        match self.receiver.try_recv() {
+            Ok(payload) => Ok(Some(payload)),
+            Err(mpsc::error::TryRecvError::Empty) => Ok(None),
+            Err(mpsc::error::TryRecvError::Disconnected) => Err(Error::Disconnected),
+        }
+    }
+
+    /// Receive a notification asynchronously
+    pub async fn recv(&mut self) -> Option<NotificationPayload> {
+        self.receiver.recv().await
+    }
+}
+
+impl Drop for ActiveSubscription {
+    fn drop(&mut self) {
+        if let Some(sub_id) = self.sub_id.take() {
+            let _ = self.on_drop_notif.try_send(sub_id);
+        }
+    }
+}

crates/cdk/src/wallet/subscription/mod.rs

@@ -1,353 +0,0 @@
-//! Client for subscriptions
-//!
-//! Mint servers can send notifications to clients about changes in the state,
-//! according to NUT-17, using the WebSocket protocol. This module provides a
-//! subscription manager that allows clients to subscribe to notifications from
-//! multiple mint servers using WebSocket or with a poll-based system, using
-//! the HTTP client.
-use std::collections::HashMap;
-use std::fmt::Debug;
-use std::sync::Arc;
-
-use cdk_common::subscription::{Params, SubId};
-use tokio::sync::{mpsc, RwLock};
-use tokio::task::JoinHandle;
-use tracing::error;
-#[cfg(target_arch = "wasm32")]
-use wasm_bindgen_futures;
-
-use super::Wallet;
-use crate::mint_url::MintUrl;
-use crate::wallet::MintConnector;
-
-mod http;
-#[cfg(all(
-    not(feature = "http_subscription"),
-    feature = "mint",
-    not(target_arch = "wasm32")
-))]
-mod ws;
-
-type WsSubscriptionBody = (mpsc::Sender<NotificationPayload>, Params);
-
-/// Subscription manager
-///
-/// This structure should be instantiated once per wallet at most. It is
-/// cloneable since all its members are Arcs.
-///
-/// The main goal is to provide a single interface to manage multiple
-/// subscriptions to many servers to subscribe to events. If supported, the
-/// WebSocket method is used to subscribe to server-side events. Otherwise, a
-/// poll-based system is used, where a background task fetches information about
-/// the resource every few seconds and notifies subscribers of any change
-/// upstream.
-///
-/// The subscribers have a simple-to-use interface, receiving an
-/// ActiveSubscription struct, which can be used to receive updates and to
-/// unsubscribe from updates automatically on the drop.
-#[derive(Debug, Clone)]
-pub struct SubscriptionManager {
-    all_connections: Arc<RwLock<HashMap<MintUrl, SubscriptionClient>>>,
-    http_client: Arc<dyn MintConnector + Send + Sync>,
-    prefer_http: bool,
-}
-
-impl SubscriptionManager {
-    /// Create a new subscription manager
-    pub fn new(http_client: Arc<dyn MintConnector + Send + Sync>, prefer_http: bool) -> Self {
-        Self {
-            all_connections: Arc::new(RwLock::new(HashMap::new())),
-            http_client,
-            prefer_http,
-        }
-    }
-
-    /// Subscribe to updates from a mint server with a given filter
-    pub async fn subscribe(
-        &self,
-        mint_url: MintUrl,
-        filter: Params,
-        wallet: Arc<Wallet>,
-    ) -> ActiveSubscription {
-        let subscription_clients = self.all_connections.read().await;
-        let id = filter.id.clone();
-        if let Some(subscription_client) = subscription_clients.get(&mint_url) {
-            let (on_drop_notif, receiver) = subscription_client.subscribe(filter).await;
-            ActiveSubscription::new(receiver, id, on_drop_notif)
-        } else {
-            drop(subscription_clients);
-
-            #[cfg(all(
-                not(feature = "http_subscription"),
-                feature = "mint",
-                not(target_arch = "wasm32")
-            ))]
-            let is_ws_support = self
-                .http_client
-                .get_mint_info()
-                .await
-                .map(|info| !info.nuts.nut17.supported.is_empty())
-                .unwrap_or_default();
-
-            #[cfg(any(
-                feature = "http_subscription",
-                not(feature = "mint"),
-                target_arch = "wasm32"
-            ))]
-            let is_ws_support = false;
-
-            let is_ws_support = if self.prefer_http {
-                false
-            } else {
-                is_ws_support
-            };
-
-            tracing::debug!(
-                "Connect to {:?} to subscribe. WebSocket is supported ({})",
-                mint_url,
-                is_ws_support
-            );
-
-            let mut subscription_clients = self.all_connections.write().await;
-            let subscription_client = SubscriptionClient::new(
-                mint_url.clone(),
-                self.http_client.clone(),
-                is_ws_support,
-                wallet,
-            );
-            let (on_drop_notif, receiver) = subscription_client.subscribe(filter).await;
-            subscription_clients.insert(mint_url, subscription_client);
-
-            ActiveSubscription::new(receiver, id, on_drop_notif)
-        }
-    }
-}
-
-/// Subscription client
-///
-/// If the server supports WebSocket subscriptions, this client will be used,
-/// otherwise the HTTP pool and pause will be used (which is the less efficient
-/// method).
-#[derive(Debug)]
-pub struct SubscriptionClient {
-    new_subscription_notif: mpsc::Sender<SubId>,
-    on_drop_notif: mpsc::Sender<SubId>,
-    subscriptions: Arc<RwLock<HashMap<SubId, WsSubscriptionBody>>>,
-    worker: Option<JoinHandle<()>>,
-}
-
-type NotificationPayload = crate::nuts::NotificationPayload<String>;
-
-/// Active Subscription
-pub struct ActiveSubscription {
-    sub_id: Option<SubId>,
-    on_drop_notif: mpsc::Sender<SubId>,
-    receiver: mpsc::Receiver<NotificationPayload>,
-}
-
-impl ActiveSubscription {
-    fn new(
-        receiver: mpsc::Receiver<NotificationPayload>,
-        sub_id: SubId,
-        on_drop_notif: mpsc::Sender<SubId>,
-    ) -> Self {
-        Self {
-            sub_id: Some(sub_id),
-            on_drop_notif,
-            receiver,
-        }
-    }
-
-    /// Try to receive a notification
-    pub fn try_recv(&mut self) -> Result<Option<NotificationPayload>, Error> {
-        match self.receiver.try_recv() {
-            Ok(payload) => Ok(Some(payload)),
-            Err(mpsc::error::TryRecvError::Empty) => Ok(None),
-            Err(mpsc::error::TryRecvError::Disconnected) => Err(Error::Disconnected),
-        }
-    }
-
-    /// Receive a notification asynchronously
-    pub async fn recv(&mut self) -> Option<NotificationPayload> {
-        self.receiver.recv().await
-    }
-}
-
-impl Drop for ActiveSubscription {
-    fn drop(&mut self) {
-        if let Some(sub_id) = self.sub_id.take() {
-            let _ = self.on_drop_notif.try_send(sub_id);
-        }
-    }
-}
-
-/// Subscription client error
-#[derive(thiserror::Error, Debug)]
-pub enum Error {
-    /// Url error
-    #[error("Could not join paths: {0}")]
-    Url(#[from] crate::mint_url::Error),
-    /// Disconnected from the notification channel
-    #[error("Disconnected from the notification channel")]
-    Disconnected,
-}
-
-impl SubscriptionClient {
-    /// Create new [`SubscriptionClient`]
-    pub fn new(
-        url: MintUrl,
-        http_client: Arc<dyn MintConnector + Send + Sync>,
-        prefer_ws_method: bool,
-        wallet: Arc<Wallet>,
-    ) -> Self {
-        let subscriptions = Arc::new(RwLock::new(HashMap::new()));
-        let (new_subscription_notif, new_subscription_recv) = mpsc::channel(100);
-        let (on_drop_notif, on_drop_recv) = mpsc::channel(1000);
-
-        Self {
-            new_subscription_notif,
-            on_drop_notif,
-            subscriptions: subscriptions.clone(),
-            worker: Self::start_worker(
-                prefer_ws_method,
-                http_client,
-                url,
-                subscriptions,
-                new_subscription_recv,
-                on_drop_recv,
-                wallet,
-            ),
-        }
-    }
-
-    #[allow(unused_variables)]
-    fn start_worker(
-        prefer_ws_method: bool,
-        http_client: Arc<dyn MintConnector + Send + Sync>,
-        url: MintUrl,
-        subscriptions: Arc<RwLock<HashMap<SubId, WsSubscriptionBody>>>,
-        new_subscription_recv: mpsc::Receiver<SubId>,
-        on_drop_recv: mpsc::Receiver<SubId>,
-        wallet: Arc<Wallet>,
-    ) -> Option<JoinHandle<()>> {
-        #[cfg(any(
-            feature = "http_subscription",
-            not(feature = "mint"),
-            target_arch = "wasm32"
-        ))]
-        return Self::http_worker(
-            http_client,
-            subscriptions,
-            new_subscription_recv,
-            on_drop_recv,
-            wallet,
-        );
-
-        #[cfg(all(
-            not(feature = "http_subscription"),
-            feature = "mint",
-            not(target_arch = "wasm32")
-        ))]
-        if prefer_ws_method {
-            Self::ws_worker(
-                http_client,
-                url,
-                subscriptions,
-                new_subscription_recv,
-                on_drop_recv,
-                wallet,
-            )
-        } else {
-            Self::http_worker(
-                http_client,
-                subscriptions,
-                new_subscription_recv,
-                on_drop_recv,
-                wallet,
-            )
-        }
-    }
-
-    /// Subscribe to a WebSocket channel
-    pub async fn subscribe(
-        &self,
-        filter: Params,
-    ) -> (mpsc::Sender<SubId>, mpsc::Receiver<NotificationPayload>) {
-        let mut subscriptions = self.subscriptions.write().await;
-        let id = filter.id.clone();
-
-        let (sender, receiver) = mpsc::channel(10_000);
-        subscriptions.insert(id.clone(), (sender, filter));
-        drop(subscriptions);
-
-        let _ = self.new_subscription_notif.send(id).await;
-        (self.on_drop_notif.clone(), receiver)
-    }
-
-    /// HTTP subscription client
-    ///
-    /// This is a poll based subscription, where the client will poll the server
-    /// from time to time to get updates, notifying the subscribers on changes
-    fn http_worker(
-        http_client: Arc<dyn MintConnector + Send + Sync>,
-        subscriptions: Arc<RwLock<HashMap<SubId, WsSubscriptionBody>>>,
-        new_subscription_recv: mpsc::Receiver<SubId>,
-        on_drop: mpsc::Receiver<SubId>,
-        wallet: Arc<Wallet>,
-    ) -> Option<JoinHandle<()>> {
-        let http_worker = http::http_main(
-            vec![],
-            http_client,
-            subscriptions,
-            new_subscription_recv,
-            on_drop,
-            wallet,
-        );
-
-        #[cfg(target_arch = "wasm32")]
-        {
-            wasm_bindgen_futures::spawn_local(http_worker);
-            None
-        }
-
-        #[cfg(not(target_arch = "wasm32"))]
-        {
-            Some(tokio::spawn(http_worker))
-        }
-    }
-
-    /// WebSocket subscription client
-    ///
-    /// This is a WebSocket based subscription, where the client will connect to
-    /// the server and stay there idle waiting for server-side notifications
-    #[cfg(all(
-        not(feature = "http_subscription"),
-        feature = "mint",
-        not(target_arch = "wasm32")
-    ))]
-    fn ws_worker(
-        http_client: Arc<dyn MintConnector + Send + Sync>,
-        url: MintUrl,
-        subscriptions: Arc<RwLock<HashMap<SubId, WsSubscriptionBody>>>,
-        new_subscription_recv: mpsc::Receiver<SubId>,
-        on_drop: mpsc::Receiver<SubId>,
-        wallet: Arc<Wallet>,
-    ) -> Option<JoinHandle<()>> {
-        Some(tokio::spawn(ws::ws_main(
-            http_client,
-            url,
-            subscriptions,
-            new_subscription_recv,
-            on_drop,
-            wallet,
-        )))
-    }
-}
-
-impl Drop for SubscriptionClient {
-    fn drop(&mut self) {
-        if let Some(handle) = self.worker.take() {
-            handle.abort();
-        }
-    }
-}