wip

Cargo.toml

@@ -17,9 +17,8 @@ async-std = { version = "1.1.0", features=["unstable","attributes"] }
 crossbeam = "0.7.3"
 log = "0.4.8"
 env_logger = "0.7.1"
-serde = "1.0.104"
+serde = { version = "1.0.104", features = ["derive"] }
 serde_json = { version = "1.0.41", features=["preserve_order","arbitrary_precision"] }
-serde_derive = "1.0.104"
 dirs = "2.0"
 futures = "0.3.1"
 snap = "0.2.5"

examples/ssb-cli.rs

@@ -69,7 +69,7 @@ async fn main() -> AnyResult<()> {
     println!("💃 handshake complete");
 
     let (box_stream_read, box_stream_write) =
-        BoxStream::from_handhake(&mut socket, handshake, 0x8000)
+        BoxStream::from_handshake(&socket,&socket,handshake, 0x8000)
         .split_read_write();
 
     let mut client = ApiClient::new(RpcClient::new(box_stream_read, box_stream_write));

src/patchwork/api.rs

@@ -2,7 +2,7 @@ use async_std::io::{Read, Write};
 use serde_json;
 use std::str::FromStr;
 
-use crate::rpc::{RpcClient, Header, RequestNo, RpcType};
+use crate::rpc::{RpcClient, Header, RequestNo, RpcType, Body};
 use crate::feed::Message;
 use crate::feed::Feed;
 
@@ -273,15 +273,24 @@ impl<R: Read + Unpin, W: Write + Unpin> ApiClient<R, W> {
     // whoami: sync
     // Get information about the current ssb-server user.
     pub async fn send_whoami(&mut self) -> Result<RequestNo> {
-        let args: [&str; 0] = [];
-        let req_no = self.rpc.send(&["whoami"], RpcType::Async, &args).await?;
+        let body = Body {
+            name : vec!["whoami".to_string()],
+            rpc_type : RpcType::Async,
+            args : Vec::<String>::new()
+        };
+        let req_no = self.rpc.send(&body).await?;
         Ok(req_no)
     }
 
     // get: async
     // Get a message by its hash-id. (sould start with %)
     pub async fn send_get(&mut self, msg_id: &str) -> Result<RequestNo> {
-        let req_no = self.rpc.send(&["get"], RpcType::Async, &msg_id).await?;
+        let body = Body {
+            name : vec!["get".to_string()],
+            rpc_type : RpcType::Async,
+            args : vec![msg_id.to_string()]
+        };
+        let req_no = self.rpc.send(&body).await?;
         Ok(req_no)
     }
     // createHistoryStream: source
@@ -290,10 +299,12 @@ impl<R: Read + Unpin, W: Write + Unpin> ApiClient<R, W> {
         &mut self,
         args: &'a CreateHistoryStreamArgs<'a>,
     ) -> Result<RequestNo> {
-        let req_no = self
-            .rpc
-            .send(&["createHistoryStream"], RpcType::Source, &args)
-            .await?;
+        let body = Body {
+            name : vec!["createHistoryStream".to_string()],
+            rpc_type : RpcType::Source,
+            args : args
+        };
+        let req_no = self.rpc.send(&body).await?;
         Ok(req_no)
     }
 
@@ -303,18 +314,24 @@ impl<R: Read + Unpin, W: Write + Unpin> ApiClient<R, W> {
         &mut self,
         args: &'a CreateStreamArgs<u64>,
     ) -> Result<RequestNo> {
-        let req_no = self
-            .rpc
-            .send(&["createFeedStream"], RpcType::Source, &args)
-            .await?;
+        let body = Body {
+            name : vec!["createFeedStream".to_string()],
+            rpc_type : RpcType::Source,
+            args
+        };
+        let req_no = self.rpc.send(&body).await?;
         Ok(req_no)
     }
 
     // latest: source
     // Get the seq numbers of the latest messages of all users in the database.
     pub async fn send_latest(&mut self) -> Result<RequestNo> {
-        let args: [&str; 0] = [];
-        let req_no = self.rpc.send(&["latest"], RpcType::Source, &args).await?;
+        let body = Body {
+            name : vec!["latest".to_string()],
+            rpc_type : RpcType::Source,
+            args : Vec::<String>::new(),
+        };
+        let req_no = self.rpc.send(&body).await?;
         Ok(req_no)
     }
 }

src/rpc/client.rs

@@ -24,9 +24,35 @@ pub enum BodyType {
     JSON,
 }
 
-#[derive(Debug,PartialEq)]
+/*
+    let mut body = String::from("{\"name\":");
+    body.push_str(&serde_json::to_string(&name)?);
+    body.push_str(",\"type\":\"");
+    body.push_str(rpc_type.rpc_id());
+    body.push_str("\",\"args\":[");
+    body.push_str(&serde_json::to_string(&args)?);
+    body.push_str("]}");
+*/
+
+#[derive(Serialize)]
+pub struct Body<T:serde::Serialize> {
+    pub name : Vec<String>,
+    #[serde(rename="type")]
+    pub rpc_type : RpcType,
+    pub args : T,
+}
+
+impl<T:serde::Serialize> Body<T> {
+    pub fn new(name: Vec<String>, rpc_type : RpcType, args:T) -> Self {
+        Body { name, rpc_type, args }
+    }
+}
+
+#[derive(Serialize,Debug,PartialEq)]
 pub enum RpcType {
+    #[serde(rename="async")]
     Async,
+    #[serde(rename="source")]
     Source,
 }
 
@@ -38,7 +64,6 @@ impl RpcType {
         }
     }
 }
-
 #[derive(Debug,PartialEq)]
 pub struct Header {
     pub req_no : RequestNo,
@@ -124,28 +149,24 @@ impl<R:io::Read+Unpin , W:io::Write+Unpin> RpcClient<R,W> {
         Ok((rpc_header,rpc_body))
     }
 
-    pub async fn send<T:serde::Serialize>(&mut self, name : &[&str], rpc_type: RpcType, args :&T) -> Result<RequestNo>{
+    pub async fn send<T:serde::Serialize>(&mut self, body : &Body<T>) -> Result<RequestNo>{
 
         self.req_no+=1;
 
-        let mut body = String::from("{\"name\":");
-        body.push_str(&serde_json::to_string(&name)?);
-        body.push_str(",\"type\":\"");
-        body.push_str(rpc_type.rpc_id());
-        body.push_str("\",\"args\":[");
-        body.push_str(&serde_json::to_string(&args)?);
-        body.push_str("]}");
+        let body_str = serde_json::to_string(body)?;
 
         let rpc_header = Header {
             req_no : self.req_no,
-            is_stream : rpc_type == RpcType::Source,
+            is_stream : body.rpc_type == RpcType::Source,
             is_end_or_error : false,
             body_type : BodyType::JSON,
-            body_len : body.len() as u32,
+            body_len : body_str.as_bytes().len() as u32,
         }.to_array();
 
+        println!("\n{}\n",body_str);
+
         self.box_writer.write_all(&rpc_header[..]).await?;
-        self.box_writer.write_all(body.as_bytes()).await?;
+        self.box_writer.write_all(body_str.as_bytes()).await?;
         self.box_writer.flush().await?;
 
         Ok(self.req_no)

src/rpc/mod.rs

@@ -1,5 +1,5 @@
 mod client;
 mod error;
 
-pub use client::{RpcClient,Header, RequestNo, RpcType};
+pub use client::{RpcClient,Header, RequestNo, RpcType, Body};
 pub use error::{Error,Result};