[ws server]: fix bug in subscription response.

`.await` was missing in RegisteredSubscription::send() and no responses
were actually sent which this commit fixes.

Author: niklasad1

src/ws/server.rs

@@ -238,13 +238,13 @@ impl Server {
         subscribe_method_name: String,
         unsubscribe_method_name: String,
     ) -> Result<RegisteredSubscription, ()> {
+        log::debug!("[register_subscription]: subscribe_method={}, unsubscribe_method={}", subscribe_method_name, unsubscribe_method_name);
         {
             let mut registered_methods = self.registered_methods.lock();
             if !registered_methods.insert(subscribe_method_name.clone()) {
                 return Err(());
             }
             if !registered_methods.insert(unsubscribe_method_name.clone()) {
-                registered_methods.remove(&subscribe_method_name);
                 return Err(());
             }
         }
@@ -253,13 +253,13 @@ impl Server {
             .next_subscription_unique_id
             .fetch_add(1, atomic::Ordering::Relaxed);
 
-        let _ = self
+        self
             .to_back
             .unbounded_send(FrontToBack::RegisterSubscription {
                 unique_id,
                 subscribe_method: subscribe_method_name,
                 unsubscribe_method: unsubscribe_method_name,
-            });
+            }).map_err(|_| ())?;
 
         Ok(RegisteredSubscription {
             to_back: self.to_back.clone(),
@@ -299,11 +299,12 @@ impl RegisteredMethod {
 
 impl RegisteredSubscription {
     /// Sends out a value to all the registered clients.
+    // TODO: return `Result<(), ()>`
     pub async fn send(&mut self, value: JsonValue) {
         let _ = self.to_back.send(FrontToBack::SendOutNotif {
             unique_id: self.unique_id,
             notification: value,
-        });
+        }).await;
     }
 }
 
@@ -314,6 +315,7 @@ impl IncomingRequest {
     }
 
     /// Respond to the request.
+    // TODO: return `Result<(), ()>`
     pub async fn respond(mut self, response: impl Into<Result<JsonValue, common::Error>>) {
         let _ = self
             .to_back
@@ -381,6 +383,7 @@ async fn background_task(
                 subscribe_method,
                 unsubscribe_method,
             })) => {
+                log::debug!("server register subscription=id={:?}, subscribe_method:{}, unsubscribe_method={}", unique_id, subscribe_method, unsubscribe_method);
                 debug_assert_ne!(subscribe_method, unsubscribe_method);
                 debug_assert!(!subscribe_methods.contains_key(&subscribe_method));
                 debug_assert!(!subscribe_methods.contains_key(&unsubscribe_method));
@@ -399,15 +402,19 @@ async fn background_task(
                 unique_id,
                 notification,
             })) => {
+                log::debug!("server send response to subscription={:?}", unique_id);
                 debug_assert!(subscribed_clients.contains_key(&unique_id));
                 if let Some(clients) = subscribed_clients.get(&unique_id) {
+                    log::debug!("{} client(s) is subscribing to {:?}", clients.len(), unique_id);
                     for client in clients {
                         debug_assert_eq!(active_subscriptions.get(client), Some(&unique_id));
                         debug_assert!(server.subscription_by_id(*client).is_some());
                         if let Some(sub) = server.subscription_by_id(*client) {
                             sub.push(notification.clone()).await;
                         }
                     }
+                } else {
+                    log::warn!("server subscription: {:?} not found", unique_id);
                 }
             }
             Either::Right(RawServerEvent::Notification(notification)) => {