Merge pull request #38 from mycognosist/ebt_exploration

Add RPC helpers for EBT

Cargo.toml

@@ -10,14 +10,14 @@ license = "AGPL-3.0"
 name = "kuska_ssb"
 
 [dependencies]
-kuska-handshake = { version="0.2", features=["sync","async_std"] }
+kuska-handshake = { version = "0.2", features=["async_std"] }
 kuska-sodiumoxide = "0.2.5-0"
 base64 = "0.11.0"
 hex = "0.4.0"
-async-std = { version = "1.5.0", features=["unstable","attributes"] }
+async-std = { version = "1.12.0", features=["unstable", "attributes"] }
 log = "0.4.8"
 serde = { version = "1.0.104", features = ["derive"] }
-serde_json = { version = "1.0.48", features=["preserve_order","arbitrary_precision"] }
+serde_json = { version = "1.0.48", features=["preserve_order", "arbitrary_precision"] }
 dirs = "2.0"
 futures = "0.3.4"
 get_if_addrs = "0.5.3"

README.md

@@ -1,11 +1,10 @@
 [![Build Status](https://github.com/kuska-ssb/kuska-ssb/workflows/Rust/badge.svg)](https://github.com/kuska-ssb/kuska-ssb/actions?query=workflow%3ARust)
 
-
 # kuska-ssb
 Secure Scuttlebutt library
 
 kuska means _together_ in [Runasimi](https://en.wikipedia.org/wiki/Quechuan_languages)
 
 kuska is an implementation of decentralized social network [Secure Scuttlebutt](https://scuttlebutt.nz/) written in rust, it does not aim to provide a user interface and the functionality implemented in some clients like [Patchwork](https://github.com/ssbc/patchwork), [Patchbay](https://github.com/ssbc/patchbay), but the full set of libraries to be able to develop applications for the secure scuttlebutt network.
 
-kuska-ssb is the implementation of protocols involved in ssb (unless the handhake and box stream that are in the [ssb-handshake repo](https://github.com/Kuska-ssb/kuska-handshake)), detailed information about the protocol can be found in https://ssbc.github.io/scuttlebutt-protocol-guide/ and https://scuttlebot.io/apis/scuttlebot/ssb.html
+kuska-ssb is the implementation of protocols involved in ssb (excluding the handshake and box stream which are in the [ssb-handshake repo](https://github.com/Kuska-ssb/kuska-handshake)), detailed information about the protocol can be found in https://ssbc.github.io/scuttlebutt-protocol-guide/ and https://scuttlebot.io/apis/scuttlebot/ssb.html

examples/ssb-cli.rs

@@ -116,7 +116,7 @@ where
                 RecvMsg::ErrorResponse(message) => {
                     return std::result::Result::Err(Box::new(AppError::new(message)));
                 }
-                RecvMsg::CancelStreamRespose() => break,
+                RecvMsg::CancelStreamResponse() => break,
                 _ => {}
             }
         }

src/api/dto/blobs.rs

@@ -1,4 +1,4 @@
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BlobsGetIn {
     // key : ID of the blob. Required.
     pub key: String,
@@ -7,7 +7,7 @@ pub struct BlobsGetIn {
     // If the blob is not exactly this size then reject the request. Optional.
     pub size: Option<u64>,
 
-    // max 	Maximum size of the blob in bytes. If the blob is larger then reject
+    // max : Maximum size of the blob in bytes. If the blob is larger then reject
     // the request. Only makes sense to specify max if you don’t already know size. Optional.
     pub max: Option<u64>,
 }

src/api/dto/ebt.rs

@@ -0,0 +1,14 @@
+#[derive(Debug, Serialize, Deserialize)]
+pub struct EbtReplicate {
+    pub version: u16,
+    pub format: String,
+}
+
+impl Default for EbtReplicate {
+    fn default() -> Self {
+        Self {
+            version: 3,
+            format: "classic".to_string(),
+        }
+    }
+}

src/api/dto/mod.rs

@@ -1,5 +1,6 @@
 mod blobs;
 pub mod content;
+mod ebt;
 mod error;
 mod history_stream;
 mod latest;
@@ -8,6 +9,7 @@ mod tangles;
 mod whoami;
 
 pub use blobs::*;
+pub use ebt::*;
 pub use error::*;
 pub use history_stream::*;
 pub use latest::*;

src/api/helper.rs

@@ -18,6 +18,7 @@ pub enum ApiMethod {
     BlobsGet,
     CreateFeedStream,
     CreateHistoryStream,
+    EbtReplicate,
     FriendsBlocks,
     FriendsHops,
     FriendsIsFollowing,
@@ -44,6 +45,7 @@ impl ApiMethod {
             BlobsGet => &["blobs", "get"],
             CreateFeedStream => &["createFeedStream"],
             CreateHistoryStream => &["createHistoryStream"],
+            EbtReplicate => &["ebt", "replicate"],
             FriendsBlocks => &["friends", "blocks"],
             FriendsHops => &["friends", "hops"],
             FriendsIsBlocking => &["friends", "isBlocking"],
@@ -69,6 +71,7 @@ impl ApiMethod {
             ["blobs", "get"] => Some(BlobsGet),
             ["createFeedStream"] => Some(CreateFeedStream),
             ["createHistoryStream"] => Some(CreateHistoryStream),
+            ["ebt", "replicate"] => Some(EbtReplicate),
             ["friends", "blocks"] => Some(FriendsBlocks),
             ["friends", "hops"] => Some(FriendsHops),
             ["friends", "isBlocking"] => Some(FriendsIsBlocking),
@@ -198,6 +201,37 @@ impl<W: Write + Unpin> ApiCaller<W> {
         Ok(req_no)
     }
 
+    /// Send ["ebt", "replicate"] request.
+    pub async fn ebt_replicate_req_send(&mut self, args: &dto::EbtReplicate) -> Result<RequestNo> {
+        let req_no = self
+            .rpc
+            .send_request(
+                ApiMethod::EbtReplicate.selector(),
+                RpcType::Duplex,
+                ArgType::Array,
+                &args,
+                &None::<()>,
+            )
+            .await?;
+        Ok(req_no)
+    }
+
+    /// Send EBT vector clock (aka. notes) response.
+    pub async fn ebt_clock_res_send(&mut self, req_no: RequestNo, clock: &str) -> Result<()> {
+        self.rpc
+            .send_response(req_no, RpcType::Duplex, BodyType::JSON, clock.as_bytes())
+            .await?;
+        Ok(())
+    }
+
+    /// Send feed response as part of EBT.
+    pub async fn ebt_feed_res_send(&mut self, req_no: RequestNo, feed: &str) -> Result<()> {
+        self.rpc
+            .send_response(req_no, RpcType::Duplex, BodyType::JSON, feed.as_bytes())
+            .await?;
+        Ok(())
+    }
+
     /// Send feed response.
     pub async fn feed_res_send(&mut self, req_no: RequestNo, feed: &str) -> Result<()> {
         self.rpc

src/rpc/stream.rs

@@ -155,7 +155,7 @@ pub enum RecvMsg {
     RpcResponse(BodyType, Vec<u8>),
     OtherRequest(BodyType, Vec<u8>),
     ErrorResponse(String),
-    CancelStreamRespose(),
+    CancelStreamResponse(),
 }
 
 impl<R: io::Read + Unpin> RpcReader<R> {
@@ -186,7 +186,7 @@ impl<R: io::Read + Unpin> RpcReader<R> {
             }
         } else if rpc_header.is_end_or_error {
             if rpc_header.is_stream {
-                Ok((-rpc_header.req_no, RecvMsg::CancelStreamRespose()))
+                Ok((-rpc_header.req_no, RecvMsg::CancelStreamResponse()))
             } else {
                 let err: ErrorMessage = serde_json::from_slice(&body_raw)?;
                 Ok((