feat(event streaming): implement push-only streaming channels and SSE (…

…#1945) This implements streaming channels using mpsc(underlying part of SSE) and SSE for sending data to clients continuously, NETWORK event is implemented to show the new functionality. All platforms other than WASM are supported. --------- Signed-off-by: onur-ozkan <work@onurozkan.dev>
KomodoPlatform · Sep 28, 2023 · e9559c2 · e9559c2
1 parent 2ce6e4a
commit e9559c2
Show file tree

Hide file tree

Showing 23 changed files with 659 additions and 78 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,44 +1,45 @@
 [workspace]
 members = [
+    "mm2src/coins_activation",
     "mm2src/coins",
     "mm2src/coins/utxo_signer",
-    "mm2src/coins_activation",
     "mm2src/common/shared_ref_counter",
     "mm2src/crypto",
     "mm2src/db_common",
     "mm2src/derives/enum_from",
-    "mm2src/derives/ser_error",
     "mm2src/derives/ser_error_derive",
+    "mm2src/derives/ser_error",
     "mm2src/floodsub",
     "mm2src/gossipsub",
-    "mm2src/mm2_gui_storage",
     "mm2src/hw_common",
     "mm2src/mm2_bin_lib",
-    "mm2src/mm2_bitcoin/crypto",
     "mm2src/mm2_bitcoin/chain",
+    "mm2src/mm2_bitcoin/crypto",
     "mm2src/mm2_bitcoin/keys",
-    "mm2src/mm2_bitcoin/rpc",
     "mm2src/mm2_bitcoin/primitives",
+    "mm2src/mm2_bitcoin/rpc",
     "mm2src/mm2_bitcoin/script",
-    "mm2src/mm2_bitcoin/serialization",
     "mm2src/mm2_bitcoin/serialization_derive",
+    "mm2src/mm2_bitcoin/serialization",
     "mm2src/mm2_bitcoin/test_helpers",
     "mm2src/mm2_core",
     "mm2src/mm2_db",
     "mm2src/mm2_err_handle",
     "mm2src/mm2_eth",
+    "mm2src/mm2_event_stream",
     "mm2src/mm2_git",
+    "mm2src/mm2_gui_storage",
     "mm2src/mm2_io",
     "mm2src/mm2_libp2p",
+    "mm2src/mm2_main",
     "mm2src/mm2_metamask",
     "mm2src/mm2_metrics",
-    "mm2src/mm2_main",
     "mm2src/mm2_net",
     "mm2src/mm2_number",
     "mm2src/mm2_rpc",
     "mm2src/mm2_state_machine",
-    "mm2src/rpc_task",
     "mm2src/mm2_test_helpers",
+    "mm2src/rpc_task",
     "mm2src/trezor",
 ]
 

diff --git a/examples/sse/README.md b/examples/sse/README.md
@@ -0,0 +1,14 @@
+# Listening event-stream from komodo-defi-framework
+
+1. Start komodo-defi-framework with event streaming activated
+2. Run a local HTTP server
+    - if you use Python 3, run:
+   ```
+   python3 -m http.server 8000
+   ```
+    - if you use Python 2, run:
+   ```
+   python -m SimpleHTTPServer 8000
+   ```
+
+You should now be able to observe events from the komodo-defi-framework through the SSE.
diff --git a/examples/sse/index.html b/examples/sse/index.html
@@ -0,0 +1,26 @@
+<!DOCTYPE html>
+<html>
+
+<body>
+
+  <h1>Events</h1>
+  <div id="result"></div>
+
+  <script>
+    if (typeof (EventSource) !== "undefined") {
+      var source = new EventSource("http://localhost:7783/event-stream", {
+        withCredentials: false,
+      });
+      source.onmessage = function (event) {
+        var currentDatetime = new Date().toLocaleString();
+        var eventData = currentDatetime + ": " + event.data + "<hr/>";
+        document.getElementById("result").insertAdjacentHTML("afterbegin", eventData);
+      };
+    } else {
+      document.getElementById("result").innerHTML = "Sorry, your browser does not support server-sent events...";
+    }
+  </script>
+
+</body>
+
+</html>
diff --git a/examples/wasm/README.md b/examples/wasm/README.md
@@ -15,4 +15,4 @@ via [WebAssembly](https://developer.mozilla.org/en-US/docs/WebAssembly)
    ```
    Read more about
    [running a simple local HTTP server](https://developer.mozilla.org/en-US/docs/Learn/Common_questions/set_up_a_local_testing_server#running_a_simple_local_http_server)
-3. Open webpage in your browser http://localhost:8000/wasm_build/index.html
+3. Open webpage in your browser http://localhost:8000/wasm_build/index.html
diff --git a/examples/wasm/index.html b/examples/wasm/index.html
@@ -1,27 +1,32 @@
 <!DOCTYPE html>
 <html lang="en">
+
 <head>
     <meta charset="UTF-8">
     <title>MM2 example</title>
 </head>
+
 <body>
 
-<div>
-    <label for="wid_conf_input">Config JSON</label>
-    <br>
-    <textarea id="wid_conf_input" cols="100" rows="7">{"gui":"WASMTEST","mm2":1,"passphrase":"wasmtest","allow_weak_password":true,"rpc_password":"testpsw","netid":7777,"coins":[{"coin":"ETH","protocol":{"type":"ETH"}},{"coin":"RICK","overwintered":1,"txversion":4,"protocol":{"type":"UTXO"}},{"coin":"MORTY","overwintered":1,"txversion":4,"protocol":{"type":"UTXO"}}]}</textarea>
-    <br>
-    <button id="wid_run_mm2_button" disabled="disabled">Start MM2</button>
-    <button id="wid_stop_mm2_button" disabled="disabled">Stop MM2</button>
-    <br>
-    <br>
-    <label for="wid_rpc_input">RPC payload JSON</label>
-    <br>
-    <textarea id="wid_rpc_input" cols="100" rows="9">[{"userpass":"testpsw","method":"electrum","mm2":1,"coin":"RICK","tx_history":true,"servers":[{"url":"electrum1.cipig.net:30017","protocol":"WSS"}]},{"userpass":"testpsw","method":"electrum","mm2":1,"coin":"MORTY","tx_history":true,"servers":[{"url":"electrum1.cipig.net:30018","protocol":"WSS"}]},{"userpass":"testpsw","method":"enable","mm2":1,"coin":"ETH","swap_contract_address":"0x8500AFc0bc5214728082163326C2FF0C73f4a871","urls":["http://eth1.cipig.net:8555"]}]</textarea>
-    <br>
-    <button id="wid_mm2_rpc_button" disabled="disabled">Submit request</button>
-</div>
+    <div>
+        <label for="wid_conf_input">Config JSON</label>
+        <br>
+        <textarea id="wid_conf_input" cols="100"
+            rows="7">{"gui":"WASMTEST","mm2":1,"passphrase":"wasmtest","allow_weak_password":true,"rpc_password":"testpsw","netid":7777,"coins":[{"coin":"ETH","protocol":{"type":"ETH"}},{"coin":"RICK","overwintered":1,"txversion":4,"protocol":{"type":"UTXO"}},{"coin":"MORTY","overwintered":1,"txversion":4,"protocol":{"type":"UTXO"}}]}</textarea>
+        <br>
+        <button id="wid_run_mm2_button" disabled="disabled">Start MM2</button>
+        <button id="wid_stop_mm2_button" disabled="disabled">Stop MM2</button>
+        <br>
+        <br>
+        <label for="wid_rpc_input">RPC payload JSON</label>
+        <br>
+        <textarea id="wid_rpc_input" cols="100"
+            rows="9">[{"userpass":"testpsw","method":"electrum","mm2":1,"coin":"RICK","tx_history":true,"servers":[{"url":"electrum1.cipig.net:30017","protocol":"WSS"}]},{"userpass":"testpsw","method":"electrum","mm2":1,"coin":"MORTY","tx_history":true,"servers":[{"url":"electrum1.cipig.net:30018","protocol":"WSS"}]},{"userpass":"testpsw","method":"enable","mm2":1,"coin":"ETH","swap_contract_address":"0x8500AFc0bc5214728082163326C2FF0C73f4a871","urls":["http://eth1.cipig.net:8555"]}]</textarea>
+        <br>
+        <button id="wid_mm2_rpc_button" disabled="disabled">Submit request</button>
+    </div>
 
 </body>
 <script src="script.js" type="module"></script>
-</html>
+
+</html>
diff --git a/mm2src/mm2_core/Cargo.toml b/mm2src/mm2_core/Cargo.toml
@@ -7,15 +7,16 @@ edition = "2021"
 doctest = false
 
 [dependencies]
-async-trait = "0.1"
 arrayref = "0.3"
+async-trait = "0.1"
 cfg-if = "1.0"
 common = { path = "../common" }
 db_common = { path = "../db_common" }
 derive_more = "0.99"
 futures = { version = "0.3", package = "futures", features = ["compat", "async-await", "thread-pool"] }
 hex = "0.4.2"
 lazy_static = "1.4"
+mm2_event_stream = { path = "../mm2_event_stream" }
 mm2_metrics = { path = "../mm2_metrics" }
 primitives = { path = "../mm2_bitcoin/primitives" }
 rand = { version = "0.7", features = ["std", "small_rng", "wasm-bindgen"] }

diff --git a/mm2src/mm2_core/src/mm_ctx.rs b/mm2src/mm2_core/src/mm_ctx.rs
@@ -6,6 +6,7 @@ use common::log::{self, LogLevel, LogOnError, LogState};
 use common::{cfg_native, cfg_wasm32, small_rng};
 use gstuff::{try_s, Constructible, ERR, ERRL};
 use lazy_static::lazy_static;
+use mm2_event_stream::{controller::Controller, Event, EventStreamConfiguration};
 use mm2_metrics::{MetricsArc, MetricsOps};
 use primitives::hash::H160;
 use rand::Rng;
@@ -72,6 +73,10 @@ pub struct MmCtx {
     pub initialized: Constructible<bool>,
     /// True if the RPC HTTP server was started.
     pub rpc_started: Constructible<bool>,
+    /// Controller for continuously streaming data using streaming channels of `mm2_event_stream`.
+    pub stream_channel_controller: Controller<Event>,
+    /// Configuration of event streaming used for SSE.
+    pub event_stream_configuration: Option<EventStreamConfiguration>,
     /// True if the MarketMaker instance needs to stop.
     pub stop: Constructible<bool>,
     /// Unique context identifier, allowing us to more easily pass the context through the FFI boundaries.  
@@ -133,6 +138,8 @@ impl MmCtx {
             metrics: MetricsArc::new(),
             initialized: Constructible::default(),
             rpc_started: Constructible::default(),
+            stream_channel_controller: Controller::new(),
+            event_stream_configuration: None,
             stop: Constructible::default(),
             ffi_handle: Constructible::default(),
             ordermatch_ctx: Mutex::new(None),
@@ -680,8 +687,17 @@ impl MmCtxBuilder {
         let mut ctx = MmCtx::with_log_state(log);
         ctx.mm_version = self.version;
         ctx.datetime = self.datetime;
+
         if let Some(conf) = self.conf {
-            ctx.conf = conf
+            ctx.conf = conf;
+
+            let event_stream_configuration = &ctx.conf["event_stream_configuration"];
+            if !event_stream_configuration.is_null() {
+                let event_stream_configuration: EventStreamConfiguration =
+                    json::from_value(event_stream_configuration.clone())
+                        .expect("Invalid json value in 'event_stream_configuration'.");
+                ctx.event_stream_configuration = Some(event_stream_configuration);
+            }
         }
 
         #[cfg(target_arch = "wasm32")]

diff --git a/mm2src/mm2_event_stream/Cargo.toml b/mm2src/mm2_event_stream/Cargo.toml
@@ -0,0 +1,18 @@
+[package]
+name = "mm2_event_stream"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+async-trait = "0.1"
+cfg-if = "1.0"
+common = { path = "../common" }
+parking_lot = "0.12"
+serde = { version = "1", features = ["derive", "rc"] }
+tokio = { version = "1", features = ["sync"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["sync", "macros", "time", "rt"] }
+
+[target.'cfg(target_arch = "wasm32")'.dependencies]
+wasm-bindgen-test = { version = "0.3.2" }
diff --git a/mm2src/mm2_event_stream/src/behaviour.rs b/mm2src/mm2_event_stream/src/behaviour.rs
@@ -0,0 +1,15 @@
+use crate::EventStreamConfiguration;
+use async_trait::async_trait;
+
+#[async_trait]
+pub trait EventBehaviour {
+    /// Unique name of the event.
+    const EVENT_NAME: &'static str;
+
+    /// Event handler that is responsible for broadcasting event data to the streaming channels.
+    async fn handle(self, interval: f64);
+
+    /// Spawns the `Self::handle` in a separate thread if the event is active according to the mm2 configuration.
+    /// Does nothing if the event is not active.
+    fn spawn_if_active(self, config: &EventStreamConfiguration);
+}