Add subscriptions support

light-node/src/commands/start.rs

@@ -123,7 +123,7 @@ impl config::Override<LightNodeConfig> for StartCmd {
  * trusted state and store it in the store ...
  * TODO: this should take traits ... but how to deal with the State ?
  * TODO: better name ?
-*/
+ */
 async fn subjective_init(
     height: Height,
     vals_hash: Hash,

tendermint/Cargo.toml

@@ -56,8 +56,9 @@ thiserror = "1"
 toml = { version = "0.5" }
 uuid = { version = "0.8", default-features = false }
 zeroize = { version = "1.1", features = ["zeroize_derive"] }
+async-tungstenite = {version="0.4", features = ["tokio-runtime"]}
+tokio = { version = "0.2", features = ["macros"] }
 ripemd160 = "0.8"
 
 [dev-dependencies]
 serde_json = "1"
-tokio = { version = "0.2", features = ["macros"] }

tendermint/src/block/commit_sig.rs

@@ -4,7 +4,8 @@ use crate::serializers;
 use crate::{account, Signature, Time};
 use serde::{de::Error as _, Deserialize, Deserializer, Serialize, Serializer};
 
-/// BlockIDFlag is used to indicate the validator has voted either for nil, a particular BlockID or was absent.
+/// BlockIDFlag is used to indicate the validator has voted either for nil, a particular BlockID or
+/// was absent.
 #[derive(Copy, Clone, Debug, PartialEq)]
 pub enum BlockIDFlag {
     /// BlockIDFlagAbsent - no vote was received from a validator.

tendermint/src/chain/id.rs

@@ -71,7 +71,7 @@ impl FromStr for Id {
 
         for byte in name.as_bytes() {
             match byte {
-                b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' => (),
+                b'a'..=b'z' | b'A'..=b'Z' | b'0'..=b'9' | b'-' | b'_' | b'.' => (),
                 _ => return Err(Kind::Parse.into()),
             }
         }

tendermint/src/rpc.rs

@@ -5,6 +5,7 @@
 mod client;
 pub mod endpoint;
 pub mod error;
+pub mod event_listener;
 mod id;
 mod method;
 pub mod request;

tendermint/src/rpc/client.rs

@@ -4,7 +4,7 @@ use crate::{
     abci::{self, Transaction},
     block::Height,
     net,
-    rpc::{self, endpoint::*, Error, Response},
+    rpc::{endpoint::*, Error, Request, Response},
     Genesis,
 };
 use bytes::buf::ext::BufExt;
@@ -153,7 +153,7 @@ impl Client {
     /// Perform a request against the RPC endpoint
     pub async fn perform<R>(&self, request: R) -> Result<R::Response, Error>
     where
-        R: rpc::Request,
+        R: Request,
     {
         let request_body = request.into_json();
 
@@ -182,7 +182,6 @@ impl Client {
                     .unwrap(),
             );
         }
-
         let http_client = hyper::Client::builder().build_http();
         let response = http_client.request(request).await?;
         let response_body = hyper::body::aggregate(response.into_body()).await?;

tendermint/src/rpc/endpoint.rs

@@ -11,4 +11,5 @@ pub mod genesis;
 pub mod health;
 pub mod net_info;
 pub mod status;
+pub mod subscribe;
 pub mod validators;

tendermint/src/rpc/endpoint/subscribe.rs

@@ -0,0 +1,46 @@
+//! `/subscribe` endpoint JSONRPC wrapper
+
+use crate::rpc;
+use serde::{Deserialize, Serialize};
+use std::io::Read;
+
+/// Subscribe request for events on websocket
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct Request {
+    query: String,
+}
+
+impl Request {
+    /// Query the Tendermint nodes event and stream events over web socket
+    pub fn new(query: String) -> Self {
+        Self { query }
+    }
+}
+
+impl rpc::Request for Request {
+    type Response = Response;
+
+    fn method(&self) -> rpc::Method {
+        rpc::Method::Subscribe
+    }
+}
+
+/// Status responses
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct Response {}
+
+/// Subcribe is weird RPC endpoint. It's only meaningful at websocket response and there isn't a
+/// synchronous reponse offered. It there is an error it's asynchronous and we don't try and stich
+/// the async response back together with the request.
+impl rpc::Response for Response {
+    /// We throw away response data JSON string so swallow errors and return the empty Response
+    fn from_string(_response: impl AsRef<[u8]>) -> Result<Self, rpc::Error> {
+        Ok(Response {})
+    }
+
+    /// We throw away responses in `subscribe` to swallow errors from the `io::Reader` and provide
+    /// the Response
+    fn from_reader(_reader: impl Read) -> Result<Self, rpc::Error> {
+        Ok(Response {})
+    }
+}

tendermint/src/rpc/error.rs

@@ -1,5 +1,6 @@
 //! JSONRPC error types
 
+use async_tungstenite::tungstenite::Error as WSError;
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use std::fmt::{self, Display};
 use thiserror::Error;
@@ -44,6 +45,11 @@ impl Error {
         Error::new(Code::InvalidParams, Some(data.to_string()))
     }
 
+    /// Create a new websocket error
+    pub fn websocket_error(cause: impl Into<String>) -> Error {
+        Error::new(Code::WebSocketError, Some(cause.into()))
+    }
+
     /// Create a new method-not-found error
     pub fn method_not_found(name: &str) -> Error {
         Error::new(Code::MethodNotFound, Some(name.to_string()))
@@ -108,16 +114,26 @@ impl From<hyper::Error> for Error {
     }
 }
 
+impl From<WSError> for Error {
+    fn from(websocket_error: WSError) -> Error {
+        Error::websocket_error(websocket_error.to_string())
+    }
+}
+
 /// Tendermint RPC error codes.
 ///
 /// See `func RPC*Error()` definitions in:
-/// <https://github.com/tendermint/tendermint/blob/master/rpc/lib/types/types.go>
+/// <https://github.com/tendermint/tendermint/blob/master/rpc/jsonrpc/types/types.go>
 #[derive(Copy, Clone, Debug, Eq, Error, Hash, PartialEq, PartialOrd, Ord)]
 pub enum Code {
     /// Low-level HTTP error
     #[error("HTTP error")]
     HttpError,
 
+    /// Low-level Websocket error
+    #[error("Websocket Error")]
+    WebSocketError,
+
     /// Parse error i.e. invalid JSON (-32700)
     #[error("Parse error. Invalid JSON")]
     ParseError,
@@ -158,6 +174,7 @@ impl From<i32> for Code {
     fn from(value: i32) -> Code {
         match value {
             0 => Code::HttpError,
+            1 => Code::WebSocketError,
             -32700 => Code::ParseError,
             -32600 => Code::InvalidRequest,
             -32601 => Code::MethodNotFound,
@@ -173,6 +190,7 @@ impl From<Code> for i32 {
     fn from(code: Code) -> i32 {
         match code {
             Code::HttpError => 0,
+            Code::WebSocketError => 1,
             Code::ParseError => -32700,
             Code::InvalidRequest => -32600,
             Code::MethodNotFound => -32601,