feat: domain request method for TransactorClient

src/services/core/classes.rs

@@ -0,0 +1 @@
+pub type OperationDomain = String;

src/services/core/mod.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 //
 
+pub mod classes;
 pub(crate) mod ser;
+pub mod storage;
 
 use crate::services::transactor::tx::Doc;
 use serde::{Deserialize, Serialize};

src/services/core/storage.rs

@@ -0,0 +1,15 @@
+use super::classes::OperationDomain;
+use serde::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct DomainResult<T> {
+    pub domain: OperationDomain,
+    pub value: T,
+}
+
+impl<T: PartialEq> PartialEq for DomainResult<T> {
+    fn eq(&self, other: &Self) -> bool {
+        self.domain.eq(&other.domain) && self.value.eq(&other.value)
+    }
+}

src/services/transactor/backend/http.rs

@@ -1,5 +1,8 @@
 use crate::Result;
 use crate::services::core::WorkspaceUuid;
+use crate::services::core::classes::OperationDomain;
+use crate::services::core::storage::DomainResult;
+use crate::services::transactor::backend::Backend;
 use crate::services::transactor::methods::Method;
 use crate::services::{JsonClient, TokenProvider};
 use reqwest_middleware::ClientWithMiddleware;
@@ -40,6 +43,30 @@ impl HttpBackend {
             }),
         }
     }
+
+    pub(crate) async fn get_path<T: DeserializeOwned + Send>(
+        &self,
+        path: &str,
+        params: impl IntoIterator<Item = (String, Value)>,
+    ) -> Result<T> {
+        let mut url = self.base().join(path)?;
+        let mut qp = url.query_pairs_mut();
+        for (name, value) in params {
+            qp.append_pair(&name, &value.to_string());
+        }
+        drop(qp);
+
+        <crate::services::HttpClient as JsonClient>::get(&self.inner.client, self, url).await
+    }
+
+    pub(crate) async fn post_path<T: DeserializeOwned + Send, Q: Serialize>(
+        &self,
+        path: &str,
+        body: &Q,
+    ) -> Result<T> {
+        let url = self.base().join(path)?;
+        <crate::services::HttpClient as JsonClient>::post(&self.inner.client, self, url, body).await
+    }
 }
 
 impl JsonClient for HttpBackend {
@@ -73,51 +100,52 @@ impl TokenProvider for &'_ HttpBackend {
     }
 }
 
-trait HttpMethod {
-    fn path(&self, workspace: WorkspaceUuid) -> String;
-}
-
-impl HttpMethod for Method {
-    fn path(&self, workspace: WorkspaceUuid) -> String {
-        format!("/api/v1/{}/{}", self.kebab(), workspace)
-    }
-}
-
 impl super::Backend for HttpBackend {
     async fn get<T: DeserializeOwned + Send>(
         &self,
         method: Method,
         params: impl IntoIterator<Item = (String, Value)>,
     ) -> Result<T> {
-        let url = {
-            let mut url = self.base().join(&method.path(self.inner.workspace))?;
-            let mut qp = url.query_pairs_mut();
-
-            for (name, value) in params {
-                match value {
-                    Value::String(s) => {
-                        qp.append_pair(&name, &s);
-                    }
-                    _ => {
-                        qp.append_pair(&name, &value.to_string());
-                    }
-                }
-            }
-            drop(qp);
-
-            url
-        };
-
-        <crate::services::HttpClient as JsonClient>::get(&self.inner.client, self, url).await
+        self.get_path(
+            &format!("/api/v1/{}/{}", method.kebab(), self.workspace()),
+            params,
+        )
+        .await
     }
 
     async fn post<T: DeserializeOwned + Send, Q: Serialize>(
         &self,
         method: Method,
         body: &Q,
     ) -> Result<T> {
-        let url = self.base().join(&method.path(self.inner.workspace))?;
-        <crate::services::HttpClient as JsonClient>::post(&self.inner.client, self, url, body).await
+        self.post_path(
+            &format!("/api/v1/{}/{}", method.kebab(), self.workspace()),
+            body,
+        )
+        .await
+    }
+
+    async fn domain_request<T: DeserializeOwned + Send, Q: Serialize>(
+        &self,
+        domain: OperationDomain,
+        operation: &str,
+        params: &Q,
+    ) -> Result<DomainResult<T>> {
+        let params = (String::from("params"), serde_json::to_value(params)?);
+        self.get_path(
+            &format!(
+                "/api/v1/{}/{domain}/{operation}/{}",
+                Method::Event.kebab(),
+                self.workspace()
+            ),
+            std::iter::once(params),
+        )
+        .await
+    }
+
+    async fn tx_raw<T: Serialize, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
+        self.post_path(&format!("/api/v1/tx/{}", self.workspace()), &tx)
+            .await
     }
 
     fn base(&self) -> &Url {

src/services/transactor/backend/mod.rs

@@ -1,6 +1,9 @@
 use crate::Result;
 use crate::services::TokenProvider;
 use crate::services::core::WorkspaceUuid;
+use crate::services::core::classes::OperationDomain;
+use crate::services::core::storage::DomainResult;
+use crate::services::transactor::Transaction;
 use crate::services::transactor::methods::Method;
 use serde::Serialize;
 use serde::de::DeserializeOwned;
@@ -24,6 +27,19 @@ pub trait Backend: Clone + TokenProvider + 'static {
         body: &Q,
     ) -> Result<T>;
 
+    async fn domain_request<T: DeserializeOwned + Send, Q: Serialize>(
+        &self,
+        domain: OperationDomain,
+        operation: &str,
+        params: &Q,
+    ) -> Result<DomainResult<T>>;
+
+    async fn tx_raw<T: Serialize, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R>;
+
+    async fn tx<T: Transaction, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
+        self.tx_raw(tx.to_value()?).await
+    }
+
     fn base(&self) -> &Url;
 
     fn workspace(&self) -> WorkspaceUuid;

src/services/transactor/backend/ws.rs

@@ -1,4 +1,6 @@
 use crate::services::core::WorkspaceUuid;
+use crate::services::core::classes::OperationDomain;
+use crate::services::core::storage::DomainResult;
 use crate::services::rpc::util::OkResponse;
 use crate::services::rpc::{HelloRequest, HelloResponse, ReqId, Request, Response};
 use crate::services::transactor::backend::Backend;
@@ -85,6 +87,8 @@ async fn socket_task(
                     let id = next_id.fetch_add(1, Ordering::Relaxed);
                     payload["id"] = Value::Number(id.into());
 
+                    trace!(target: "ws", %payload, "Sending message");
+
                     pending.insert(id.into(), reply_tx);
                     write.send(encode_message(&payload, binary_mode)?).await?;
                 },
@@ -408,6 +412,33 @@ impl Backend for WsBackend {
         send_and_wait(&self.inner.cmd_tx, payload).await
     }
 
+    async fn domain_request<T: DeserializeOwned + Send, Q: Serialize>(
+        &self,
+        domain: OperationDomain,
+        operation: &str,
+        params: &Q,
+    ) -> Result<DomainResult<T>> {
+        let payload = Request {
+            id: None,
+            method: Method::Request.camel().to_string(),
+            params: vec![
+                Value::String(domain),
+                serde_json::json!({
+                    operation: {
+                        "params": params
+                    },
+                }),
+            ],
+            time: None,
+        };
+
+        send_and_wait(&self.inner.cmd_tx, payload).await
+    }
+
+    async fn tx_raw<T: Serialize, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
+        self.post(Method::Tx, &tx).await
+    }
+
     fn base(&self) -> &Url {
         &self.inner.base
     }
@@ -422,7 +453,6 @@ async fn send_and_wait<T: DeserializeOwned + Send, U: Serialize + Debug>(
     payload: Request<U>,
 ) -> Result<T> {
     let payload = serde_json::to_value(&payload)?;
-    trace!(target: "ws", %payload, "Sending message");
 
     let (reply_tx, reply_rx) = oneshot::channel();
     cmd_tx.send(Command::Call { payload, reply_tx }).ok();

src/services/transactor/methods.rs

@@ -24,6 +24,7 @@ api_methods!(
     FindAll: "find-all", "findAll",
     EnsurePerson: "ensure-person", "ensurePerson",
     Tx: "tx", "tx",
+    Request: "request", "domainRequest",
     Event: "event", "event",
     Ping: "ping", "ping",
     Hello: "hello", "hello",

src/services/transactor/mod.rs

@@ -13,12 +13,14 @@
 //
 
 use crate::Result;
+use crate::services::ForceScheme;
 use crate::services::core::WorkspaceUuid;
+use crate::services::core::classes::OperationDomain;
+use crate::services::core::storage::DomainResult;
 use crate::services::transactor::backend::Backend;
 use crate::services::transactor::backend::http::{HttpBackend, HttpClient};
 use crate::services::transactor::backend::ws::{WsBackend, WsBackendOpts};
 use crate::services::transactor::methods::Method;
-use crate::services::{ForceScheme, JsonClient};
 use secrecy::{ExposeSecret, SecretString};
 use serde::{Serialize, de::DeserializeOwned};
 use serde_json::Value;
@@ -93,6 +95,23 @@ impl<B: Backend> TransactorClient<B> {
         self.backend.post(method, body).await
     }
 
+    pub async fn domain_request<T: DeserializeOwned + Send, Q: Serialize>(
+        &self,
+        domain: OperationDomain,
+        operation: &str,
+        params: &Q,
+    ) -> Result<DomainResult<T>> {
+        self.backend.domain_request(domain, operation, params).await
+    }
+
+    pub async fn tx_raw<T: Serialize, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
+        self.backend.tx_raw(tx).await
+    }
+
+    pub async fn tx<T: Transaction, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
+        self.backend.tx(tx).await
+    }
+
     pub(in crate::services::transactor) fn backend(&self) -> &B {
         &self.backend
     }
@@ -112,19 +131,6 @@ impl TransactorClient<HttpBackend> {
     }
 }
 
-impl TransactorClient<HttpBackend> {
-    pub async fn tx_raw<T: Serialize, R: DeserializeOwned>(&self, tx: T) -> Result<R> {
-        let path = format!("/api/v1/tx/{}", self.workspace());
-        let url = self.base().join(&path)?;
-
-        <HttpBackend as JsonClient>::post(self.backend(), &self, url, &tx).await
-    }
-
-    pub async fn tx<T: Transaction, R: DeserializeOwned>(&self, tx: T) -> Result<R> {
-        self.tx_raw(tx.to_value()?).await
-    }
-}
-
 impl TransactorClient<WsBackend> {
     pub async fn new_ws(
         base: Url,
@@ -146,16 +152,6 @@ impl TransactorClient<WsBackend> {
     }
 }
 
-impl TransactorClient<WsBackend> {
-    pub async fn tx_raw<T: Serialize, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
-        self.backend.post(Method::Tx, &tx).await
-    }
-
-    pub async fn tx<T: Transaction, R: DeserializeOwned + Send>(&self, tx: T) -> Result<R> {
-        self.tx_raw(tx.to_value()?).await
-    }
-}
-
 #[cfg(feature = "kafka")]
 pub mod kafka {
     use super::*;

src/services/transactor/tx.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 //
 
+use crate::services::core::classes::OperationDomain;
 use crate::services::core::ser::Data;
 use crate::services::core::{PersonId, Ref, Timestamp};
 use crate::services::event::{Class, Event};
@@ -130,8 +131,6 @@ impl Class for TxRemoveDoc {
 
 impl Event for TxRemoveDoc {}
 
-pub type OperationDomain = String;
-
 #[derive(Serialize, Deserialize, Debug)]
 pub struct TxDomainEvent<T> {
     #[serde(flatten)]