create a POC for p2p gateway

Cargo.lock

@@ -54,7 +54,7 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47"
 dependencies = [
- "getrandom 0.2.3",
+ "getrandom 0.2.4",
  "once_cell",
  "version_check",
 ]
@@ -827,12 +827,6 @@ dependencies = [
  "tungstenite",
 ]
 
-[[package]]
-name = "cassowary"
-version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53"
-
 [[package]]
 name = "cc"
 version = "1.0.72"
@@ -1017,7 +1011,7 @@ version = "0.1.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "615f6e27d000a2bffbc7f2f6a8669179378fa27ee4d0a509e985dfc0a7defb40"
 dependencies = [
- "getrandom 0.2.3",
+ "getrandom 0.2.4",
  "lazy_static",
  "proc-macro-hack",
  "tiny-keccak",
@@ -2181,9 +2175,9 @@ dependencies = [
 
 [[package]]
 name = "getrandom"
-version = "0.2.3"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7fcd999463524c52659517fe2cea98493cfe485d10565e7b0fb07dbba7ad2753"
+checksum = "418d37c8b1d42553c93648be529cb70f920d3baf8ef469b74b9638df426e0b4c"
 dependencies = [
  "cfg-if 1.0.0",
  "libc",
@@ -2816,14 +2810,6 @@ dependencies = [
  "value-bag",
 ]
 
-[[package]]
-name = "map"
-version = "0.3.0"
-dependencies = [
- "termion",
- "tui",
-]
-
 [[package]]
 name = "matches"
 version = "0.1.9"
@@ -3662,7 +3648,7 @@ version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d34f1408f55294453790c48b2f1ebbb1c5b4b7563eb1f418bcfcfdbb06ebb4e7"
 dependencies = [
- "getrandom 0.2.3",
+ "getrandom 0.2.4",
 ]
 
 [[package]]
@@ -3846,7 +3832,7 @@ version = "0.4.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "528532f3d801c87aec9def2add9ca802fe569e44a544afe633765267840abe64"
 dependencies = [
- "getrandom 0.2.3",
+ "getrandom 0.2.4",
  "redox_syscall 0.2.10",
 ]
 
@@ -4378,9 +4364,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.7.0"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ecab6c735a6bb4139c0caafd0cc3635748bbb3acf4550e8138122099251f309"
+checksum = "f2dd574626839106c320a323308629dcb1acfc96e32a8cba364ddc61ac23ee83"
 
 [[package]]
 name = "smol"
@@ -5509,19 +5495,6 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ae2f58a822f08abdaf668897e96a5656fe72f5a9ce66422423e8849384872e6"
 
-[[package]]
-name = "tui"
-version = "0.16.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39c8ce4e27049eed97cfa363a5048b09d995e209994634a0efc26a14ab6c0c23"
-dependencies = [
- "bitflags",
- "cassowary",
- "termion",
- "unicode-segmentation",
- "unicode-width",
-]
-
 [[package]]
 name = "tungstenite"
 version = "0.16.0"
@@ -5641,7 +5614,7 @@ version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7"
 dependencies = [
- "getrandom 0.2.3",
+ "getrandom 0.2.4",
 ]
 
 [[package]]

src/node/service/gateway_p2p.rs

@@ -0,0 +1,170 @@
+use std::sync::Arc;
+use std::io;
+
+use rand::Rng;
+use async_executor::Executor;
+use log::*;
+
+use crate::{
+    blockchain::{rocks::columns, RocksColumn, Slab, SlabStore},
+    util::serial::{Decodable, Encodable},
+    Error, Result,
+    net::{P2pPtr, P2p, Settings}
+};
+
+#[allow(dead_code)]
+pub struct Gateway {
+    p2p: P2pPtr,
+    slabstore: Arc<SlabStore>,
+}
+
+impl Gateway {
+    pub fn new(
+        _settings: Settings,
+        rocks: RocksColumn<columns::Slabs>,
+    ) -> Result<Self> {
+        let slabstore = SlabStore::new(rocks)?;
+        let settings = Settings::default();
+
+        let p2p = P2p::new(settings);
+        Ok(Self { p2p, slabstore})
+    }
+
+    pub async fn start(&self, executor: Arc<Executor<'_>>) -> Result<()> {
+
+        self.p2p.clone().start(executor.clone()).await?;
+
+        self.p2p.clone().run(executor.clone()).await?;
+
+        Ok(())
+    }
+
+
+    async fn publish(&self, _msg: GatewayMessage) -> Result<()> {
+        Ok(())
+    }
+
+    async fn subscribe_loop(&self, _executor: Arc<Executor<'_>>) -> Result<()> {
+        let new_channel_sub = self.p2p.subscribe_channel().await;
+
+        loop {
+            let channel = new_channel_sub.receive().await?;
+            // do something
+        }
+
+    }
+
+    async fn handle_msg(
+        msg: GatewayMessage,
+        _slabstore: Arc<SlabStore>,
+    ) -> Result<()> {
+        match msg.get_command() {
+            GatewayCommand::PutSlab => {
+                debug!(target: "GATEWAY DAEMON", "Received putslab msg");
+            }
+            GatewayCommand::GetSlab => {
+                debug!(target: "GATEWAY DAEMON", "Received getslab msg");
+            }
+            GatewayCommand::GetLastIndex => {
+                debug!(target: "GATEWAY DAEMON","Received getlastindex msg");
+            }
+        }
+        Ok(())
+    }
+
+    pub async fn sync(&mut self) -> Result<u64> {
+        debug!(target: "GATEWAY CLIENT", "Start Syncing");
+        Ok(0)
+    }
+
+    pub async fn get_slab(&mut self, _index: u64) -> Result<Option<Slab>> {
+        debug!(target: "GATEWAY CLIENT","Get slab");
+        Ok(None)
+    }
+
+    pub async fn put_slab(&mut self, _slab: Slab) -> Result<()> {
+        debug!(target: "GATEWAY CLIENT","Put slab");
+        Ok(())
+    }
+
+    pub async fn get_last_index(&self) -> Result<u64> {
+        debug!(target: "GATEWAY CLIENT","Get last index");
+        Ok(0)
+    }
+
+    pub fn get_slabstore(&self) -> Arc<SlabStore> {
+        self.slabstore.clone()
+    }
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub enum GatewayCommand {
+    PutSlab,
+    GetSlab,
+    GetLastIndex,
+}
+
+#[derive(Debug, PartialEq, Clone)]
+pub struct GatewayMessage {
+    command: GatewayCommand,
+    id: u32,
+    payload: Vec<u8>,
+}
+
+impl GatewayMessage {
+    pub fn new(command: GatewayCommand, payload: Vec<u8>) -> Self {
+        let id = Self::gen_id();
+        Self { command, id, payload }
+    }
+    fn gen_id() -> u32 {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+
+    pub fn get_id(&self) -> u32 {
+        self.id
+    }
+
+    pub fn get_command(&self) ->  GatewayCommand{
+        self.command.clone()
+    }
+
+    pub fn get_payload(&self) -> Vec<u8> {
+        self.payload.clone()
+    }
+}
+
+impl Encodable for GatewayMessage {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += (self.command.clone() as u8).encode(&mut s)?;
+        len += self.id.encode(&mut s)?;
+        len += self.payload.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for GatewayMessage {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        let command_code: u8 =  Decodable::decode(&mut d)?;
+        let command = match  command_code {
+            0 => GatewayCommand::GetSlab,
+            1 => GatewayCommand::PutSlab,
+            2 => GatewayCommand::GetLastIndex,
+            _ => GatewayCommand::GetLastIndex,
+        };
+
+        Ok(Self {
+            command,
+            id: Decodable::decode(&mut d)?,
+            payload: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl crate::net::Message for GatewayMessage {
+    fn name() -> &'static str {
+        "reply"
+    }
+}
+

src/node/service/mod.rs

@@ -1,4 +1,5 @@
 pub mod gateway;
 pub mod reqrep;
+pub mod gateway_p2p;
 
 pub use gateway::{GatewayClient, GatewayService, GatewaySlabsSubscriber};