KomodoPlatform · shamardy · Sep 28, 2023 · Aug 23, 2023 · Aug 23, 2023 · Aug 23, 2023
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,43 +1,44 @@
 [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/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),
@@ -677,8 +684,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);
+}