chore: Initial commit for Rust backend

.gitignore

@@ -1,6 +1,8 @@
 # See https://help.github.com/ignore-files/ for more about ignoring files.
 htdocs
 
+backend/target
+
 # dependencies
 node_modules
 

backend/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "backend"
+version = "0.1.0"
+authors = ["Maciej Hirsz <maciej.hirsz@gmail.com>"]
+edition = "2018"
+
+[dependencies]
+actix = "0.8"
+actix-web = "1.0.0-rc"
+actix-web-actors = "1.0.0"
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+chrono = { version = "0.4", features = ["serde"] }
+primitive-types = { version = "0.3.0", features = ["serde"] }

backend/src/chain.rs

@@ -0,0 +1,7 @@
+use actix::prelude::*;
+
+pub struct Chain;
+
+impl Actor for Chain {
+    type Context = Context<Self>;
+}

backend/src/main.rs

@@ -0,0 +1,40 @@
+use actix::prelude::*;
+use actix_web::{web, App, HttpRequest, HttpResponse, HttpServer, Error};
+use actix_web_actors::ws;
+
+mod node_connector;
+mod node_message;
+mod chain;
+
+use node_connector::NodeConnector;
+use chain::Chain;
+
+/// Entry point for connecting nodes
+fn node_route(
+    req: HttpRequest,
+    stream: web::Payload,
+    chain: web::Data<Addr<Chain>>,
+) -> Result<HttpResponse, Error> {
+    println!("Connection!");
+
+    ws::start(
+        NodeConnector::new(chain.get_ref().clone()),
+        &req,
+        stream,
+    )
+}
+
+fn main() -> std::io::Result<()> {
+    let sys = System::new("substrate-telemetry");
+    let chain = Chain.start();
+
+    HttpServer::new(move || {
+        App::new()
+            .data(chain.clone())
+            .service(web::resource("/submit").route(web::get().to(node_route)))
+    })
+    .bind("127.0.0.1:8080")?
+    .start();
+
+    sys.run()
+}

backend/src/node_connector.rs

@@ -0,0 +1,84 @@
+use std::time::{Duration, Instant};
+
+use actix::prelude::*;
+use actix_web_actors::ws;
+use crate::chain::Chain;
+use crate::node_message::NodeMessage;
+
+/// How often heartbeat pings are sent
+const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
+/// How long before lack of client response causes a timeout
+const CLIENT_TIMEOUT: Duration = Duration::from_secs(10);
+
+
+pub struct NodeConnector {
+    /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
+    hb: Instant,
+    /// Chain actor address
+    addr: Addr<Chain>,
+}
+
+impl Actor for NodeConnector {
+    type Context = ws::WebsocketContext<Self>;
+
+    /// Method is called on actor start. We start the heartbeat process here.
+    fn started(&mut self, ctx: &mut Self::Context) {
+        self.hb(ctx);
+    }
+}
+
+impl NodeConnector {
+    pub fn new(addr: Addr<Chain>) -> Self {
+        Self {
+            hb: Instant::now(),
+            addr,
+        }
+    }
+
+    /// Send ping every 5 seconds
+    fn hb(&self, ctx: &mut <Self as Actor>::Context) {
+        ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
+            // check client heartbeats
+            if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
+                // heartbeat timed out
+                println!("NodeConnector timeout!");
+                // stop actor
+                ctx.stop();
+            } else {
+                ctx.ping("")
+            }
+        });
+    }
+}
+
+/// Handler for `ws::Message`
+impl StreamHandler<ws::Message, ws::ProtocolError> for NodeConnector {
+    fn handle(&mut self, msg: ws::Message, ctx: &mut Self::Context) {
+        // process websocket messages
+        match msg {
+            ws::Message::Ping(msg) => {
+                self.hb = Instant::now();
+                ctx.pong(&msg);
+            }
+            ws::Message::Pong(_) => {
+                self.hb = Instant::now();
+            }
+            ws::Message::Text(text) => {
+                match serde_json::from_str::<NodeMessage>(&text) {
+                    Ok(msg) => println!("GOT\t{:?}\nFROM:\t{}\n", msg, text),
+                    _ => (),
+                    // Err(err) => println!("\t{:?}\n\t{}", err, text),
+                }
+                // ctx.text(test); // echo
+            }
+            ws::Message::Binary(bin) => {
+                println!("Binary message: {} bytes", bin.len());
+                // ctx.binary(bin); // echo
+            }
+            ws::Message::Close(_) => {
+                ctx.stop();
+            }
+            ws::Message::Nop => (),
+        }
+    }
+}

backend/src/node_message.rs

@@ -0,0 +1,61 @@
+use chrono::{DateTime, Utc};
+use serde::Deserialize;
+
+pub use primitive_types::H256 as BlockHash;
+pub type BlockNumber = u64;
+
+#[derive(Deserialize, Debug)]
+pub struct NodeMessage {
+    level: Level,
+    ts: DateTime<Utc>,
+    #[serde(flatten)]
+    details: Details,
+}
+
+#[derive(Deserialize, Debug)]
+pub enum Level {
+    #[serde(rename = "INFO")]
+    Info,
+}
+
+#[derive(Deserialize, Debug)]
+#[serde(tag = "msg")]
+pub enum Details {
+    #[serde(rename = "node.start")]
+    NodeStart(BestBlock),
+    #[serde(rename = "system.connected")]
+    SystemConnected(SystemConnected),
+    #[serde(rename = "system.interval")]
+    SystemInterval(SystemInterval),
+    #[serde(rename = "block.import")]
+    BlockImport(BestBlock),
+}
+
+#[derive(Deserialize, Debug)]
+pub struct SystemConnected {
+    pub name: Box<str>,
+    pub chain: Box<str>,
+    pub implementation: Box<str>,
+    pub version: Box<str>,
+    pub config: Option<Box<str>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct SystemInterval {
+    pub txcount: u64,
+    pub peers: u64,
+    pub memory: Option<f64>,
+    pub cpu: Option<f64>,
+    pub bandwidth_upload: Option<f64>,
+    pub bandwidth_download: Option<f64>,
+    pub finalized_height: Option<BlockNumber>,
+    pub finalized_hash: Option<BlockHash>,
+    #[serde(flatten)]
+    pub best_block: BestBlock,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct BestBlock {
+    pub best: BlockHash,
+    pub height: BlockNumber,
+}