fix: add .subscribe / async support to wasm client

bindings_wasm/tests/web.rs

@@ -5,9 +5,9 @@ use prost::Message;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen_test::*;
 use xmtp_cryptography::signature::RecoverableSignature;
-use xmtp_proto::api_client::XmtpApiClient;
+use xmtp_proto::api_client::{XmtpApiClient, XmtpApiSubscription};
 use xmtp_proto::xmtp::message_api::v1::{
-    BatchQueryRequest, Envelope, PublishRequest, QueryRequest,
+    BatchQueryRequest, Envelope, PublishRequest, QueryRequest, SubscribeRequest,
 };
 
 // Only run these tests in a browser.

xmtp/src/mock_xmtp_api_client.rs

@@ -7,12 +7,14 @@ use xmtp_proto::api_client::*;
 
 pub struct MockXmtpApiSubscription {}
 
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl XmtpApiSubscription for MockXmtpApiSubscription {
     fn is_closed(&self) -> bool {
         false
     }
 
-    fn get_messages(&self) -> Vec<Envelope> {
+    async fn get_messages(&mut self) -> Vec<Envelope> {
         vec![]
     }
 

xmtp_api_grpc/src/grpc_api_helper.rs

@@ -245,12 +245,13 @@ impl Subscription {
     }
 }
 
-impl XmtpApiSubscription for Subscription {
+#[async_trait]
+impl<'a> XmtpApiSubscription for Subscription {
     fn is_closed(&self) -> bool {
         self.closed.load(Ordering::SeqCst)
     }
 
-    fn get_messages(&self) -> Vec<Envelope> {
+    async fn get_messages(&mut self) -> Vec<Envelope> {
         let mut pending = self.pending.lock().unwrap();
         let items = pending.drain(..).collect::<Vec<Envelope>>();
         items

xmtp_api_grpc/src/lib.rs

@@ -116,12 +116,12 @@ mod tests {
             tokio::time::sleep(std::time::Duration::from_millis(100)).await;
 
             // Ensure that messages appear
-            let results = stream_handler.get_messages();
+            let results = stream_handler.get_messages().await;
             println!("{}", results.len());
             assert!(results.len() == 1);
 
             // Ensure that the messages array has been cleared
-            let second_results = stream_handler.get_messages();
+            let second_results = stream_handler.get_messages().await;
             assert!(second_results.is_empty());
 
             // Ensure the is_closed status is propagated

xmtp_api_grpc_gateway/Cargo.toml

@@ -8,12 +8,19 @@ crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 async-trait = "0.1.68"
+bytes = "1.4.0"
+futures = "0.3"
+futures-util = "0.3"
 getrandom = { version = "0.2", features = ["js"] }
 hex = "0.4"
 js-sys = "0.3"
 prost = { version = "0.11", features = ["prost-derive"] }
 prost-types = "0.11"
-reqwest = { version = "0.11.20", features = ["json"] }
+reqwest = { version = "0.11.20", features = ["json", "stream"] }
+serde = { version = "1.0.160", features = ["derive"] }
+serde_json = "1.0"
+tokio = {version="1.28.1", features=["rt", "sync", "macros", "io-util"]}
+tokio-util = {version="0.7.8", features=["codec", "rt", "io-util"]}
 wasm-bindgen = "0.2.87"
 wasm-bindgen-futures = "0.4.37"
 xmtp_cryptography = { path = "../xmtp_cryptography", features = ["ws"] }