feat(zcoin): implement zcoin/pirate transport layer for WASM

mm2src/coins/z_coin/z_rpc.rs

@@ -244,7 +244,7 @@ impl ZRpcOps for LightRpcClient {
             .into_inner();
         // without Pin method get_mut is not found in current scope
         while let Some(block) = Pin::new(&mut response).get_mut().message().await? {
-            debug!("Got block {:?}", block);
+            debug!("Got block {}", block.height);
             let height = u32::try_from(block.height)
                 .map_err(|_| UpdateBlocksCacheErr::DecodeError("Block height too large".to_string()))?;
             db.insert_block(height, block.encode_to_vec())

mm2src/mm2_main/src/wasm_tests.rs

@@ -8,11 +8,11 @@ use mm2_core::mm_ctx::MmArc;
 use mm2_number::BigDecimal;
 use mm2_rpc::data::legacy::OrderbookResponse;
 use mm2_test_helpers::electrums::{doc_electrums, marty_electrums};
-use mm2_test_helpers::for_tests::{check_recent_swaps, enable_electrum_json, init_z_coin_light, init_z_coin_status,
-                                  morty_conf, pirate_conf, rick_conf, start_swaps, test_qrc20_history_impl,
-                                  wait_for_swaps_finish_and_check_status, MarketMakerIt, Mm2InitPrivKeyPolicy,
-                                  Mm2TestConf, Mm2TestConfForSwap, ARRR, MORTY, PIRATE_ELECTRUMS,
-                                  PIRATE_LIGHTWALLETD_URLS, RICK};
+use mm2_test_helpers::for_tests::{check_recent_swaps, enable_electrum_json, enable_z_coin_light, init_z_coin_light,
+                                  init_z_coin_status, morty_conf, pirate_conf, rick_conf, start_swaps,
+                                  test_qrc20_history_impl, wait_for_swaps_finish_and_check_status, MarketMakerIt,
+                                  Mm2InitPrivKeyPolicy, Mm2TestConf, Mm2TestConfForSwap, ARRR, MORTY,
+                                  PIRATE_ELECTRUMS, PIRATE_LIGHTWALLETD_URLS, RICK};
 use mm2_test_helpers::get_passphrase;
 use mm2_test_helpers::structs::{EnableCoinBalance, InitTaskResult, InitZcoinStatus, RpcV2Response,
                                 ZCoinActivationResult};
@@ -195,32 +195,6 @@ async fn trade_test_rick_and_morty() {
     .await;
 }
 
-async fn enable_z_coin_light(
-    mm: &MarketMakerIt,
-    coin: &str,
-    electrums: &[&str],
-    lightwalletd_urls: &[&str],
-    account: Option<u32>,
-) -> ZCoinActivationResult {
-    let init = init_z_coin_light(mm, coin, electrums, lightwalletd_urls, None, account).await;
-    let init: RpcV2Response<InitTaskResult> = json::from_value(init).unwrap();
-    let timeout = wait_until_sec(300);
-
-    loop {
-        if now_sec() > timeout {
-            panic!("{} initialization timed out", coin);
-        }
-        let status = init_z_coin_status(mm, init.result.task_id).await;
-        info!("Status {}", json::to_string(&status).unwrap());
-        let status: RpcV2Response<InitZcoinStatus> = json::from_value(status).unwrap();
-        match status.result {
-            InitZcoinStatus::Ok(result) => break result,
-            InitZcoinStatus::Error(e) => panic!("{} initialization error {:?}", coin, e),
-            _ => Timer::sleep(1.).await,
-        }
-    }
-}
-
 #[wasm_bindgen_test]
 async fn activate_z_coin_light() {
     register_wasm_log();

mm2src/mm2_main/tests/integration_tests_common/mod.rs

@@ -7,9 +7,9 @@ use mm2_main::mm2::{lp_main, LpMainParams};
 use mm2_rpc::data::legacy::CoinInitResponse;
 use mm2_test_helpers::electrums::{doc_electrums, marty_electrums};
 use mm2_test_helpers::for_tests::{enable_native as enable_native_impl, init_utxo_electrum, init_utxo_status,
-                                  init_z_coin_light, init_z_coin_status, MarketMakerIt};
-use mm2_test_helpers::structs::{InitTaskResult, InitUtxoStatus, InitZcoinStatus, RpcV2Response,
-                                UtxoStandardActivationResult, ZCoinActivationResult};
+                                  MarketMakerIt};
+
+use mm2_test_helpers::structs::{InitTaskResult, InitUtxoStatus, RpcV2Response, UtxoStandardActivationResult};
 use serde_json::{self as json, Value as Json};
 use std::collections::HashMap;
 use std::env::var;
@@ -83,34 +83,6 @@ pub async fn enable_coins_rick_morty_electrum(mm: &MarketMakerIt) -> HashMap<&'s
     replies
 }
 
-pub async fn enable_z_coin_light(
-    mm: &MarketMakerIt,
-    coin: &str,
-    electrums: &[&str],
-    lightwalletd_urls: &[&str],
-    starting_date: Option<u64>,
-    account: Option<u32>,
-) -> ZCoinActivationResult {
-    let init = init_z_coin_light(mm, coin, electrums, lightwalletd_urls, starting_date, account).await;
-    let init: RpcV2Response<InitTaskResult> = json::from_value(init).unwrap();
-    let timeout = wait_until_ms(600000);
-
-    loop {
-        if now_ms() > timeout {
-            panic!("{} initialization timed out", coin);
-        }
-
-        let status = init_z_coin_status(mm, init.result.task_id).await;
-        println!("Status {}", json::to_string(&status).unwrap());
-        let status: RpcV2Response<InitZcoinStatus> = json::from_value(status).unwrap();
-        match status.result {
-            InitZcoinStatus::Ok(result) => break result,
-            InitZcoinStatus::Error(e) => panic!("{} initialization error {:?}", coin, e),
-            _ => Timer::sleep(1.).await,
-        }
-    }
-}
-
 pub async fn enable_utxo_v2_electrum(
     mm: &MarketMakerIt,
     coin: &str,

mm2src/mm2_main/tests/mm2_tests/mod.rs

@@ -8,6 +8,7 @@ mod orderbook_sync_tests;
 mod tendermint_ibc_asset_tests;
 mod tendermint_tests;
 mod z_coin_tests;
+pub use z_coin_tests::activate_z_coin_light;
 
 mod zhtlc_native_reexport {
     pub use common::executor::Timer;

mm2src/mm2_main/tests/mm2_tests/orderbook_sync_tests.rs

@@ -1,14 +1,14 @@
 use crate::integration_tests_common::{enable_coins_eth_electrum, enable_coins_rick_morty_electrum, enable_electrum,
-                                      enable_electrum_json, enable_z_coin_light};
+                                      enable_electrum_json};
 use common::{block_on, log};
 use http::StatusCode;
 use mm2_main::mm2::lp_ordermatch::MIN_ORDER_KEEP_ALIVE_INTERVAL;
 use mm2_number::{BigDecimal, BigRational, MmNumber};
 use mm2_rpc::data::legacy::{AggregatedOrderbookEntry, CoinInitResponse, OrderbookResponse};
 use mm2_test_helpers::electrums::doc_electrums;
-use mm2_test_helpers::for_tests::{eth_jst_testnet_conf, eth_testnet_conf, get_passphrase, morty_conf, orderbook_v2,
-                                  rick_conf, zombie_conf, MarketMakerIt, Mm2TestConf, DOC_ELECTRUM_ADDRS,
-                                  ETH_DEV_NODES, MARTY_ELECTRUM_ADDRS, RICK, ZOMBIE_ELECTRUMS,
+use mm2_test_helpers::for_tests::{enable_z_coin_light, eth_jst_testnet_conf, eth_testnet_conf, get_passphrase,
+                                  morty_conf, orderbook_v2, rick_conf, zombie_conf, MarketMakerIt, Mm2TestConf,
+                                  DOC_ELECTRUM_ADDRS, ETH_DEV_NODES, MARTY_ELECTRUM_ADDRS, RICK, ZOMBIE_ELECTRUMS,
                                   ZOMBIE_LIGHTWALLETD_URLS, ZOMBIE_TICKER};
 use mm2_test_helpers::get_passphrase;
 use mm2_test_helpers::structs::{GetPublicKeyResult, OrderbookV2Response, RpcV2Response, SetPriceResponse};
@@ -1337,7 +1337,6 @@ fn zhtlc_orders_sync_alice_connected_before_creation() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let set_price_json = json!({
@@ -1401,7 +1400,6 @@ fn zhtlc_orders_sync_alice_connected_after_creation() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let set_price_json = json!({
@@ -1431,7 +1429,6 @@ fn zhtlc_orders_sync_alice_connected_after_creation() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let set_price_json = json!({

mm2src/mm2_main/tests/mm2_tests/z_coin_tests.rs

@@ -3,10 +3,10 @@ use common::executor::Timer;
 use common::{block_on, log, now_ms, now_sec, wait_until_ms};
 use mm2_number::BigDecimal;
 use mm2_test_helpers::electrums::doc_electrums;
-use mm2_test_helpers::for_tests::{disable_coin, init_withdraw, pirate_conf, rick_conf, send_raw_transaction,
-                                  withdraw_status, z_coin_tx_history, zombie_conf, MarketMakerIt, Mm2TestConf, ARRR,
-                                  PIRATE_ELECTRUMS, PIRATE_LIGHTWALLETD_URLS, RICK, ZOMBIE_ELECTRUMS,
-                                  ZOMBIE_LIGHTWALLETD_URLS, ZOMBIE_TICKER};
+use mm2_test_helpers::for_tests::{disable_coin, enable_z_coin_light, init_withdraw, pirate_conf, rick_conf,
+                                  send_raw_transaction, withdraw_status, z_coin_tx_history, zombie_conf,
+                                  MarketMakerIt, Mm2TestConf, ARRR, PIRATE_ELECTRUMS, PIRATE_LIGHTWALLETD_URLS, RICK,
+                                  ZOMBIE_ELECTRUMS, ZOMBIE_LIGHTWALLETD_URLS, ZOMBIE_TICKER};
 use mm2_test_helpers::structs::{EnableCoinBalance, InitTaskResult, RpcV2Response, TransactionDetails, WithdrawStatus,
                                 ZcoinHistoryRes};
 use serde_json::{self as json, json, Value as Json};
@@ -47,7 +47,7 @@ async fn withdraw(mm: &MarketMakerIt, coin: &str, to: &str, amount: &str) -> Tra
 }
 
 #[test]
-fn activate_z_coin_light() {
+pub fn activate_z_coin_light() {
     let coins = json!([zombie_conf()]);
 
     let conf = Mm2TestConf::seednode(ZOMBIE_TEST_BALANCE_SEED, &coins);
@@ -59,7 +59,6 @@ fn activate_z_coin_light() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let balance = match activation_result.wallet_balance {
@@ -82,7 +81,6 @@ fn activate_z_coin_light_with_changing_height() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let old_first_sync_block = activation_result.first_sync_block;
@@ -109,7 +107,6 @@ fn activate_z_coin_light_with_changing_height() {
         ZOMBIE_TICKER,
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
-        Some(two_days_ago),
         None,
     ));
 
@@ -143,7 +140,6 @@ fn activate_z_coin_with_hd_account() {
         ZOMBIE_TICKER,
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
-        None,
         Some(hd_account_id),
     ));
 
@@ -172,7 +168,6 @@ fn test_z_coin_tx_history() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let tx_history = block_on(z_coin_tx_history(&mm, ZOMBIE_TICKER, 5, None));
@@ -417,7 +412,6 @@ fn withdraw_z_coin_light() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     println!("{:?}", activation_result);
@@ -460,7 +454,6 @@ fn trade_rick_zombie_light() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     println!("Bob ZOMBIE activation {:?}", zombie_activation);
@@ -494,7 +487,6 @@ fn trade_rick_zombie_light() {
         ZOMBIE_ELECTRUMS,
         ZOMBIE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     println!("Alice ZOMBIE activation {:?}", zombie_activation);
@@ -552,7 +544,6 @@ fn activate_pirate_light() {
         PIRATE_ELECTRUMS,
         PIRATE_LIGHTWALLETD_URLS,
         None,
-        None,
     ));
 
     let balance = match activation_result.wallet_balance {

mm2src/mm2_net/Cargo.toml

@@ -47,11 +47,10 @@ tower-service = "0.3"
 wasm-bindgen = "0.2.86"
 wasm-bindgen-test = { version = "0.3.2" }
 wasm-bindgen-futures = "0.4.21"
-wasm-streams = "0.3.0"
 web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase",
     "IdbCursor", "IdbCursorWithValue", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore",
     "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode",
-    "IdbVersionChangeEvent", "MessageEvent","ReadableStream", "WebSocket", "Worker"] }
+    "IdbVersionChangeEvent", "MessageEvent", "ReadableStreamDefaultReader", "ReadableStream", "WebSocket", "Worker"] }
 
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]

mm2src/mm2_net/src/grpc_web.rs

@@ -98,6 +98,7 @@ pub enum PostGrpcWebErr {
     DecodeBody(String),
     EncodeBody(String),
     InvalidRequest(String),
+    BadResponse(String),
     Internal(String),
     PayloadTooShort(String),
     Status(String),

mm2src/mm2_net/src/wasm/body_stream.rs

@@ -22,19 +22,19 @@ use byteorder::{BigEndian, ByteOrder};
 use bytes::{BufMut, Bytes, BytesMut};
 use common::{APPLICATION_GRPC_WEB, APPLICATION_GRPC_WEB_PROTO, APPLICATION_GRPC_WEB_TEXT,
              APPLICATION_GRPC_WEB_TEXT_PROTO};
-use futures_util::ready;
-use futures_util::{stream::empty, Stream, TryStreamExt};
+use futures_util::{ready, stream};
+use futures_util::{stream::empty, Stream};
 use http::{header::HeaderName, HeaderMap, HeaderValue};
 use http_body::Body;
 use httparse::{Status, EMPTY_HEADER};
-use js_sys::Uint8Array;
+use js_sys::{Object, Uint8Array};
 use pin_project::pin_project;
 use std::ops::{Deref, DerefMut};
 use std::{pin::Pin,
           task::{Context, Poll}};
-use wasm_bindgen::JsCast;
-use wasm_streams::readable::IntoStream;
-use web_sys::ReadableStream;
+use wasm_bindgen::{JsCast, JsValue};
+use wasm_bindgen_futures::JsFuture;
+use web_sys::{ReadableStream, ReadableStreamDefaultReader};
 
 /// If the 8th most significant bit of a frame is `0`, it indicates data; if `1`, it indicates a trailer.
 const TRAILER_BIT: u8 = 0b10000000;
@@ -144,11 +144,14 @@ pub struct ResponseBody {
 
 impl ResponseBody {
     /// Creates a new `ResponseBody` based on a ReadableStream and content type.
-    pub(crate) fn new(body_stream: ReadableStream, content_type: &str) -> Result<Self, PostGrpcWebErr> {
-        let body_stream = wasm_streams::ReadableStream::from_raw(body_stream.unchecked_into()).into_stream();
+    pub(crate) async fn new(body_stream: ReadableStream, content_type: &str) -> Result<Self, PostGrpcWebErr> {
+        let body_stream: ReadableStreamDefaultReader = body_stream
+            .get_reader()
+            .dyn_into()
+            .map_err(|err| PostGrpcWebErr::BadResponse(format!("{err:?}")))?;
 
         Ok(Self {
-            body_stream: BodyStream::new(body_stream),
+            body_stream: BodyStream::new(body_stream).await?,
             buf: EncodedBytes::new(content_type)?,
             incomplete_data: BytesMut::new(),
             data: None,
@@ -348,15 +351,33 @@ pub struct BodyStream {
 }
 
 impl BodyStream {
-    /// Creates a new `BodyStream` based on an IntoStream.
-    pub fn new(body_stream: IntoStream<'static>) -> Self {
-        Self {
-            body_stream: Box::pin(
-                body_stream
-                    .map_ok(|js_value| Uint8Array::new(&js_value).to_vec().into())
-                    .map_err(|err| PostGrpcWebErr::InvalidRequest(format!("{err:?}"))),
-            ),
+    /// Creates a new `BodyStream` based on an `ReadableStreamDefaultReader`.
+    pub async fn new(body_stream: ReadableStreamDefaultReader) -> Result<Self, PostGrpcWebErr> {
+        let mut chunks = vec![];
+        loop {
+            let value = JsFuture::from(body_stream.read())
+                .await
+                .map_err(|err| PostGrpcWebErr::InvalidRequest(format!("{err:?}")))?;
+            let object: Object = value
+                .dyn_into()
+                .map_err(|err| PostGrpcWebErr::BadResponse(format!("{err:?}")))?;
+            let object_value = js_sys::Reflect::get(&object, &JsValue::from_str("value"))
+                .map_err(|err| PostGrpcWebErr::BadResponse(format!("{err:?}")))?;
+            let object_progress = js_sys::Reflect::get(&object, &JsValue::from_str("done"))
+                .map_err(|err| PostGrpcWebErr::BadResponse(format!("{err:?}")))?;
+            let chunk = Uint8Array::new(&object_value).to_vec();
+            chunks.extend_from_slice(&chunk);
+
+            if object_progress.as_bool().ok_or_else(|| {
+                PostGrpcWebErr::BadResponse("Expected done(bool) field in json object response".to_string())
+            })? {
+                break;
+            }
         }
+
+        Ok(Self {
+            body_stream: Box::pin(stream::once(async { Ok(Bytes::from(chunks)) })),
+        })
     }
 
     /// Creates an empty `BodyStream`.

mm2src/mm2_net/src/wasm/http.rs

@@ -331,6 +331,7 @@ impl FetchRequest {
         let builder = Response::builder().status(status_code);
         let (builder, content_type) = set_response_headers_and_content_type(builder, &js_response)?;
         let body = ResponseBody::new(resp_stream, &content_type)
+            .await
             .map_to_mm(|err| SlurpError::InvalidRequest(format!("{err:?}")))?;
 
         Ok((