fix(api): re-enable payment notifications over websockets

also fix review comments

Author: gakonst

Cargo.lock

@@ -34,7 +34,6 @@ warp = { version = "0.2", default-features = false }
 secrecy = { version = "0.6", default-features = false, features = ["serde"] }
 lazy_static = "1.4.0"
 async-trait = "0.1.22"
-mime = "0.3.16"
 
 [dev-dependencies]
 tokio = { version = "0.2.9", features = ["rt-core", "macros"] }

crates/interledger-api/Cargo.toml

@@ -28,12 +28,13 @@ impl Client {
         }
     }
 
-    pub async fn create_engine_account<T: Display + Copy + Unpin>(
+    pub async fn create_engine_account<T: Display>(
         &self,
         engine_url: Url,
         id: T,
     ) -> Result<StatusCode, reqwest::Error> {
         let mut se_url = engine_url.clone();
+        let id: String = id.to_string();
         se_url
             .path_segments_mut()
             .expect("Invalid settlement engine URL")
@@ -56,7 +57,7 @@ impl Client {
             move || {
                 client
                     .post(se_url.as_ref())
-                    .json(&json!({"id" : id.to_string()}))
+                    .json(&json!({ "id": id }))
                     .send()
                     .map_ok(move |response| response.status())
             },

crates/interledger-api/src/http_retry.rs

@@ -101,6 +101,8 @@ pub trait NodeStore: AddressStore + Clone + Send + Sync + 'static {
     /// Sets the static routes for routing
     async fn set_static_routes<R>(&self, routes: R) -> Result<(), ()>
     where
+        // The 'async_trait lifetime is used after recommendation here:
+        // https://github.com/dtolnay/async-trait/issues/8#issuecomment-514812245
         R: IntoIterator<Item = (String, Uuid)> + Send + 'async_trait;
 
     /// Sets a single static route
@@ -113,6 +115,8 @@ pub trait NodeStore: AddressStore + Clone + Send + Sync + 'static {
     /// Sets the default settlement engines to be used for the provided asset codes
     async fn set_settlement_engines(
         &self,
+        // The 'async_trait lifetime is used after recommendation here:
+        // https://github.com/dtolnay/async-trait/issues/8#issuecomment-514812245
         asset_to_url_map: impl IntoIterator<Item = (String, Url)> + Send + 'async_trait,
     ) -> Result<(), ()>;
 

crates/interledger-api/src/lib.rs

@@ -1,6 +1,6 @@
 use crate::{http_retry::Client, number_or_string, AccountDetails, AccountSettings, NodeStore};
 use bytes::Bytes;
-use futures::{future::join_all, TryFutureExt};
+use futures::{future::join_all, Future, FutureExt, StreamExt, TryFutureExt};
 use interledger_btp::{connect_to_service_account, BtpAccount, BtpOutgoingService};
 use interledger_ccp::{CcpRoutingAccount, Mode, RouteControlRequest, RoutingRelation};
 use interledger_http::{deserialize_json, error::*, HttpAccount, HttpStore};
@@ -13,7 +13,7 @@ use interledger_service::{
 use interledger_service_util::{BalanceStore, ExchangeRateStore};
 use interledger_settlement::core::types::SettlementAccount;
 use interledger_spsp::{pay, SpspResponder};
-use interledger_stream::StreamNotificationsStore;
+use interledger_stream::{PaymentNotification, StreamNotificationsStore};
 use log::{debug, error, trace};
 use secrecy::{ExposeSecret, SecretString};
 use serde::{Deserialize, Serialize};
@@ -345,20 +345,9 @@ where
         .and(warp::path::end())
         .and(warp::ws())
         .and(with_store.clone())
-        .map(|_id: Uuid, ws: warp::ws::Ws, _store: S| {
-            ws.on_upgrade(move |_ws: warp::ws::WebSocket| {
-                async {
-                    // TODO: Implement this.
-                    unimplemented!()
-                    //     let (tx, rx) = futures::channel::mpsc::unbounded::<PaymentNotification>();
-                    //     store.add_payment_notification_subscription(id, tx);
-                    //     rx.map_err(|_| -> warp::Error { unreachable!("unbounded rx never errors") })
-                    //         .map(|notification| {
-                    //             warp::ws::Message::text(serde_json::to_string(&notification).unwrap())
-                    //         })
-                    //         .map(|_| ())
-                    //         .map_err(|err| error!("Error forwarding notifications to websocket: {:?}", err))
-                }
+        .map(|id: Uuid, ws: warp::ws::Ws, store: S| {
+            ws.on_upgrade(move |ws: warp::ws::WebSocket| {
+                notify_user(ws, id, store).map(|result| result.unwrap())
             })
         })
         .boxed();
@@ -474,11 +463,37 @@ where
         .or(get_account)
         .or(get_account_balance)
         .or(put_account_settings)
-        .or(incoming_payment_notifications) // Commented out until tungenstite ws support is added
+        .or(incoming_payment_notifications)
         .or(post_payments)
         .boxed()
 }
 
+fn notify_user(
+    socket: warp::ws::WebSocket,
+    id: Uuid,
+    store: impl StreamNotificationsStore,
+) -> impl Future<Output = Result<(), ()>> {
+    let (tx, rx) = futures::channel::mpsc::unbounded::<PaymentNotification>();
+    // the client is now subscribed
+    store.add_payment_notification_subscription(id, tx);
+
+    // Anytime something is written to tx, it will reach rx
+    // and get converted to a warp::ws::Message
+    let rx = rx.map(|notification: PaymentNotification| {
+        let msg = warp::ws::Message::text(serde_json::to_string(&notification).unwrap());
+        Ok(msg)
+    });
+
+    // Then it gets forwarded to the client
+    rx.forward(socket)
+        .map(|result| {
+            if let Err(e) = result {
+                eprintln!("websocket send error: {}", e);
+            }
+        })
+        .then(futures::future::ok)
+}
+
 async fn get_address_from_parent_and_update_routes<O, A, S>(
     mut service: O,
     parent: A,