Add ABCI crate

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 
 members = [
+    "abci",
     "light-client",
     "light-node",
     "p2p",

abci/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "tendermint-abci"
+version = "0.17.1"
+authors = ["Thane Thomson <[email protected]>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[features]
+client = []
+echo-app = []
+async = [ "async-trait", "futures", "pin-project" ]
+runtime-async-std = [ "async-channel", "async-std" ]
+runtime-std = []
+runtime-tokio = [ "tokio", "tokio-util" ]
+
+[dependencies]
+bytes = "0.6"
+eyre = "0.6"
+log = "0.4"
+tendermint = { version = "0.17.1", path = "../tendermint" }
+tendermint-proto = { version = "0.17.1", path = "../proto" }
+thiserror = "1.0"
+
+async-trait = { version = "0.1", optional = true }
+async-channel = { version = "1.5", optional = true }
+async-std = { version = "1.8", features = [ "attributes" ], optional = true }
+futures = { version = "0.3", optional = true }
+pin-project = { version = "1.0", optional = true }
+tokio = { version = "0.3", features = [ "macros", "net", "rt", "sync" ], optional = true }
+tokio-util = { version = "0.5", features = [ "codec" ], optional = true }

abci/src/application.rs

@@ -0,0 +1,30 @@
+//! ABCI application-related types.
+
+#[cfg(feature = "echo-app")]
+pub mod echo;
+
+use tendermint::abci::{request, response};
+
+/// ABCI server application interface.
+pub trait Application: Send + Clone + 'static {
+    /// Request that the ABCI server echo back the same message sent to it.
+    fn echo(&self, req: request::Echo) -> response::Echo {
+        response::Echo {
+            message: req.message,
+        }
+    }
+
+    /// Receive information about the Tendermint node and respond with
+    /// information about the ABCI application.
+    fn info(&self, _req: request::Info) -> response::Info {
+        Default::default()
+    }
+
+    /// Generic handler for mapping incoming requests to responses.
+    fn handle(&self, req: request::Request) -> response::Response {
+        match req {
+            request::Request::Echo(echo) => response::Response::Echo(self.echo(echo)),
+            request::Request::Info(info) => response::Response::Info(self.info(info)),
+        }
+    }
+}

abci/src/application/echo.rs

@@ -0,0 +1,41 @@
+//! Trivial ABCI application that just implements echo functionality.
+
+use crate::Application;
+use tendermint::abci::{request, response};
+
+/// Trivial ABCI application that just implements echo functionality.
+#[derive(Clone)]
+pub struct EchoApp {
+    data: String,
+    version: String,
+    app_version: u64,
+}
+
+impl EchoApp {
+    /// Constructor.
+    pub fn new<S: AsRef<str>>(data: S, version: S, app_version: u64) -> Self {
+        Self {
+            data: data.as_ref().to_owned(),
+            version: version.as_ref().to_owned(),
+            app_version,
+        }
+    }
+}
+
+impl Default for EchoApp {
+    fn default() -> Self {
+        EchoApp::new("Echo App", "0.0.1", 1)
+    }
+}
+
+impl Application for EchoApp {
+    fn info(&self, _req: request::Info) -> response::Info {
+        response::Info {
+            data: self.data.clone(),
+            version: self.version.clone(),
+            app_version: self.app_version,
+            last_block_height: 1,
+            last_block_app_hash: vec![],
+        }
+    }
+}

abci/src/client.rs

@@ -0,0 +1,48 @@
+//! ABCI clients for interacting with ABCI servers.
+
+use crate::runtime::{ClientCodec, Runtime, TcpStream};
+use crate::{Error, Result};
+use std::convert::TryInto;
+use tendermint::abci::request::RequestInner;
+use tendermint::abci::{request, response};
+
+/// A runtime-dependent ABCI client.
+pub struct Client<Rt: Runtime> {
+    codec: Rt::ClientCodec,
+}
+
+#[cfg(feature = "async")]
+impl<Rt: Runtime> Client<Rt> {
+    /// Connect to the ABCI server at the given network address.
+    pub async fn connect<S: AsRef<str>>(addr: S) -> Result<Self> {
+        let stream = Rt::TcpStream::connect(addr.as_ref()).await?;
+        Ok(Self {
+            codec: Rt::ClientCodec::from_tcp_stream(stream),
+        })
+    }
+
+    /// Request that the ABCI server echo back the message in the given
+    /// request.
+    pub async fn echo(&mut self, req: request::Echo) -> Result<response::Echo> {
+        self.perform(req).await
+    }
+
+    /// Provide information to the ABCI server about the Tendermint node in
+    /// exchange for information about the application.
+    pub async fn info(&mut self, req: request::Info) -> Result<response::Info> {
+        self.perform(req).await
+    }
+
+    async fn perform<Req: RequestInner>(&mut self, req: Req) -> Result<Req::Response> {
+        use futures::{SinkExt, StreamExt};
+
+        self.codec.send(req.into()).await?;
+        let res: std::result::Result<Req::Response, tendermint::Error> = self
+            .codec
+            .next()
+            .await
+            .ok_or(Error::ServerStreamTerminated)??
+            .try_into();
+        Ok(res?)
+    }
+}

abci/src/codec.rs

@@ -0,0 +1,163 @@
+//! ABCI codec.
+
+use crate::{Error, Result};
+use bytes::{Buf, BufMut, BytesMut};
+use tendermint::abci::request::Request;
+use tendermint::abci::response::Response;
+use tendermint_proto::Protobuf;
+
+// The maximum number of bytes we expect in a varint. We use this to check if
+// we're encountering a decoding error for a varint.
+const MAX_VARINT_LENGTH: usize = 16;
+
+/// Tendermint Socket Protocol encoder.
+pub struct TspEncoder {}
+
+impl TspEncoder {
+    /// Encode the given request to its raw wire-level representation and store
+    /// this in the given buffer.
+    #[cfg(feature = "client")]
+    pub fn encode_request(request: Request, mut dst: &mut BytesMut) -> Result<()> {
+        encode_length_delimited(|mut b| Ok(request.encode(&mut b)?), &mut dst)
+    }
+
+    /// Encode the given response to its raw wire-level representation and
+    /// store this in the given buffer.
+    pub fn encode_response(response: Response, mut dst: &mut BytesMut) -> Result<()> {
+        encode_length_delimited(|mut b| Ok(response.encode(&mut b)?), &mut dst)
+    }
+}
+
+/// Tendermint Socket Protocol decoder.
+pub struct TspDecoder {
+    read_buf: BytesMut,
+}
+
+impl TspDecoder {
+    /// Constructor.
+    pub fn new() -> Self {
+        Self {
+            read_buf: BytesMut::new(),
+        }
+    }
+
+    /// Attempt to decode a request from the given buffer.
+    ///
+    /// Returns `Ok(None)` if we don't yet have enough data to decode a full
+    /// request.
+    pub fn decode_request(&mut self, buf: &mut BytesMut) -> Result<Option<Request>> {
+        self.read_buf.put(buf);
+        decode_length_delimited(&mut self.read_buf, |mut b| Ok(Request::decode(&mut b)?))
+    }
+
+    /// Attempt to decode a response from the given buffer.
+    ///
+    /// Returns `Ok(None)` if we don't yet have enough data to decode a full
+    /// response.
+    #[cfg(feature = "client")]
+    pub fn decode_response(&mut self, buf: &mut BytesMut) -> Result<Option<Response>> {
+        self.read_buf.put(buf);
+        decode_length_delimited(&mut self.read_buf, |mut b| Ok(Response::decode(&mut b)?))
+    }
+}
+
+// encode_varint and decode_varint will be removed once
+// https://github.com/tendermint/tendermint/issues/5783 lands in Tendermint.
+fn encode_varint<B: BufMut>(val: u64, mut buf: &mut B) {
+    tendermint_proto::encode_varint(val << 1, &mut buf);
+}
+
+fn decode_varint<B: Buf>(mut buf: &mut B) -> Result<u64> {
+    let len = tendermint_proto::decode_varint(&mut buf)
+        .map_err(|_| Error::Protobuf(tendermint_proto::Kind::DecodeMessage.into()))?;
+    Ok(len >> 1)
+}
+
+// Allows us to avoid having to re-export `prost::Message`.
+// TODO(thane): Investigate a better approach here.
+fn encode_length_delimited<F, B>(mut encode_fn: F, mut dst: &mut B) -> Result<()>
+where
+    F: FnMut(&mut BytesMut) -> Result<()>,
+    B: BufMut,
+{
+    let mut buf = BytesMut::new();
+    encode_fn(&mut buf)?;
+    let buf = buf.freeze();
+    encode_varint(buf.len() as u64, &mut dst);
+    dst.put(buf);
+    Ok(())
+}
+
+fn decode_length_delimited<F, T>(src: &mut BytesMut, mut decode_fn: F) -> Result<Option<T>>
+where
+    F: FnMut(&mut BytesMut) -> Result<T>,
+{
+    let src_len = src.len();
+    let mut tmp = src.clone().freeze();
+    let encoded_len = match decode_varint(&mut tmp) {
+        Ok(len) => len,
+        Err(e) => {
+            return if src_len <= MAX_VARINT_LENGTH {
+                // We've potentially only received a partial length delimiter
+                Ok(None)
+            } else {
+                Err(e)
+            };
+        }
+    };
+    let remaining = tmp.remaining() as u64;
+    if remaining < encoded_len {
+        // We don't have enough data yet to decode the entire message
+        Ok(None)
+    } else {
+        let delim_len = src_len - tmp.remaining();
+        // We only advance the source buffer once we're sure we have enough
+        // data to try to decode the result.
+        src.advance(delim_len + (encoded_len as usize));
+
+        let mut result_bytes = BytesMut::from(tmp.split_to(encoded_len as usize).as_ref());
+        Ok(Some(decode_fn(&mut result_bytes)?))
+    }
+}
+
+#[cfg(feature = "client")]
+#[cfg(test)]
+mod test {
+    use super::*;
+    use tendermint::abci::request::Echo;
+
+    #[test]
+    fn single_request() {
+        let request = Request::Echo(Echo {
+            message: "Hello TSP!".to_owned(),
+        });
+        let mut buf = BytesMut::new();
+        TspEncoder::encode_request(request.clone(), &mut buf).unwrap();
+
+        let mut decoder = TspDecoder::new();
+        let decoded_request = decoder.decode_request(&mut buf).unwrap().unwrap();
+
+        assert_eq!(request, decoded_request);
+    }
+
+    #[test]
+    fn multiple_requests() {
+        let requests = (0..5)
+            .map(|r| {
+                Request::Echo(Echo {
+                    message: format!("Request {}", r),
+                })
+            })
+            .collect::<Vec<Request>>();
+        let mut buf = BytesMut::new();
+        requests
+            .iter()
+            .for_each(|request| TspEncoder::encode_request(request.clone(), &mut buf).unwrap());
+
+        let mut decoder = TspDecoder::new();
+        for request in requests {
+            let decoded = decoder.decode_request(&mut buf).unwrap().unwrap();
+            assert_eq!(decoded, request);
+        }
+    }
+}

abci/src/lib.rs

@@ -0,0 +1,38 @@
+//! ABCI framework for building applications with Tendermint.
+
+mod application;
+#[cfg(any(
+    feature = "client",
+    feature = "runtime-tokio",
+    feature = "runtime-async-std"
+))]
+mod codec;
+mod result;
+pub mod runtime;
+mod server;
+
+// Client exports
+#[cfg(feature = "client")]
+mod client;
+
+#[cfg(feature = "client")]
+pub use client::Client;
+
+#[cfg(all(feature = "client", feature = "runtime-tokio"))]
+pub type TokioClient = Client<runtime::tokio::Tokio>;
+#[cfg(all(feature = "client", feature = "runtime-async-std"))]
+pub type AsyncStdClient = Client<runtime::async_std::AsyncStd>;
+
+// Example applications
+#[cfg(feature = "echo-app")]
+pub use application::echo::EchoApp;
+
+// Common exports
+pub use application::Application;
+pub use result::{Error, Result};
+pub use server::Server;
+
+#[cfg(feature = "runtime-tokio")]
+pub type TokioServer<A> = Server<A, runtime::tokio::Tokio>;
+#[cfg(feature = "runtime-async-std")]
+pub type AsyncStdServer<A> = Server<A, runtime::async_std::AsyncStd>;