move notifier API to V1

jormungandr/src/rest/mod.rs

@@ -2,7 +2,7 @@
 
 pub mod context;
 pub mod explorer;
-pub mod notifier;
+// pub mod notifier;
 pub mod v0;
 mod v1;
 
@@ -34,12 +34,11 @@ pub async fn start_rest_server(config: Rest, explorer_enabled: bool, context: Co
     let api =
         warp::path!("api" / ..).and(v0::filter(context.clone()).or(v1::filter(context.clone())));
 
-    let notifier = notifier::filter(context.clone());
     if explorer_enabled {
         let explorer = explorer::filter(context);
-        setup_cors(api.or(notifier).or(explorer), config, stopper_rx).await;
+        setup_cors(api.or(explorer), config, stopper_rx).await;
     } else {
-        setup_cors(api.or(notifier), config, stopper_rx).await;
+        setup_cors(api, config, stopper_rx).await;
     }
 }
 

jormungandr/src/rest/v1/handlers.rs

@@ -1,4 +1,4 @@
-use crate::rest::{v1::logic, ContextLock};
+use crate::rest::{context::Error, v1::logic, ContextLock};
 use warp::{reject::Reject, Rejection, Reply};
 
 impl Reject for logic::Error {}
@@ -42,3 +42,18 @@ pub async fn get_fragments_logs(context: ContextLock) -> Result<impl Reply, Reje
         .map_err(warp::reject::custom)
         .map(|r| warp::reply::json(&r))
 }
+
+pub async fn handle_subscription(
+    ws: warp::ws::Ws,
+    context: ContextLock,
+) -> Result<impl Reply, Rejection> {
+    let context = context.read().await;
+    logic::handle_subscription(ws, &context)
+        .await
+        .map_err(warp::reject::custom)
+    // let full_context = context.try_full().map_err(warp::reject::custom)?;
+
+    // let notifier: crate::notifier::Notifier = full_context.notifier.clone();
+
+    // Ok(ws.on_upgrade(move |socket| add_connection(notifier, socket)))
+}

jormungandr/src/rest/v1/logic.rs

@@ -103,3 +103,17 @@ pub async fn get_fragments_logs(context: &Context) -> Result<Vec<FragmentLog>, E
         })?;
     reply_future.await.map_err(Into::into)
 }
+
+pub async fn handle_subscription(
+    ws: warp::ws::Ws,
+    context: &Context,
+) -> Result<impl warp::Reply, Error> {
+    let full_context = context.try_full()?;
+    let notifier: crate::notifier::Notifier = full_context.notifier.clone();
+
+    Ok(ws.on_upgrade(move |socket| add_connection(notifier, socket)))
+}
+
+async fn add_connection(notifier: crate::notifier::Notifier, socket: warp::ws::WebSocket) {
+    notifier.new_connection(socket).await;
+}

jormungandr/src/rest/v1/mod.rs

@@ -37,7 +37,13 @@ pub fn filter(
         root.and(post.or(status).or(logs)).boxed()
     };
 
-    let routes = fragments;
+    let notifier = warp::path!("notifier")
+        .and(warp::ws())
+        .and(with_context)
+        .and_then(handlers::handle_subscription)
+        .boxed();
+
+    let routes = fragments.or(notifier);
 
     root.and(routes).recover(handle_rejection).boxed()
 }

testing/jormungandr-integration-tests/src/common/jormungandr/notifier.rs

@@ -17,7 +17,7 @@ pub enum NotifierError {
 }
 
 pub fn uri_from_socket_addr(addr: SocketAddr) -> Url {
-    Url::parse(&format!("ws://{}/notifier", addr)).unwrap()
+    Url::parse(&format!("ws://{}/api/v1/notifier", addr)).unwrap()
 }
 
 /// Specialized rest api