Merge crate abci-rs into abci

CHANGELOG.md

@@ -1,5 +1,15 @@
 # CHANGELOG
 
+_TBD_
+
+## v0.8.0
+
+- Restructures `abci` to use latest `async`/`await` functionality (new design is based on https://github.com/devashishdxt/abci-rs,
+  read documentation for more information)
+- [Issue #30](https://github.com/tendermint/rust-abci/issues/30): Adds support for unix sockets
+- [Issue #107](https://github.com/tendermint/rust-abci/issues/107): Updates `tokio` to `v0.2`
+- Adds support for `async-std` executor to drive `Future`s.
+
 _March 3, 2020_
 
 ## v0.7.0

Cargo.toml

@@ -1,25 +1,38 @@
 [package]
 name = "abci"
-version = "0.7.0"
+version = "0.8.0"
 authors = ["Adrian Brink <adrian@brink-holdings.com>", "Jackson Lewis <st.japa6@gmail.com>", "Dave Bryson", "Tomas Tauber"]
 edition = "2018"
 license = "MIT/Apache-2.0"
 description = "Tendermint ABCI server for Rust"
 homepage = "https://tendermint.com/docs/spec/abci/"
 repository = "https://github.com/tendermint/rust-abci"
+categories = ["network-programming"]
 keywords = ["abci", "tendermint", "blockchain", "rust"]
 readme = "README.md"
 include = ["src/**/*", "Cargo.toml"]
 
+[package.metadata.docs.rs]
+features = ["doc"]
+rustdoc-args = ["--cfg", "feature=\"doc\""]
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
 [dependencies]
-bytes = "0.4"
-protobuf = "= 2.10.2"
-byteorder = "1.3.2"
-integer-encoding = "1.0.5"
-log = "0.4.8"
-env_logger = "0.7.0"
-tokio = { version = "0.1", default-features = false, features = ["codec", "io", "tcp", "rt-full"] }
-futures = "0.3"
+log = "0.4"
+protobuf = "2.10"
+integer-encoding = "1.0"
+async-trait = "0.1"
+async-std = { version = "1.5", optional = true }
+tokio = { version = "0.2", optional = true, features = ["io-util", "sync", "tcp", "stream", "rt-core", "uds"] }
+
+[dev-dependencies]
+env_logger = "0.7"
+tokio = { version = "0.2", features = ["macros"] }
 
 [build-dependencies]
-protobuf-codegen-pure = "= 2.10.2"
+protobuf-codegen-pure = "2.10"
+
+[features]
+default = ["tokio"]
+doc = []

README.md

@@ -15,26 +15,38 @@ applications for [Tendermint](https://github.com/tendermint/tendermint/).
 - Tendermint 0.33.1
 - ABCI 0.16.1
 
-## Installation
+## Usage
 
-### Dependencies
+Add `abci` in your `Cargo.toml`'s `dependencies` section:
 
-Make sure that you have Rust and Cargo installed. The easiest way is to follow the instructions on [rustup](https://rustup.rs/).
+```toml
+[dependencies]
+abci = "0.8"
+```
 
-To test the examples, please clone this repository.
+Each ABCI application has to implement three core traits corresponding to all three ABCI connections, `Consensus`,
+`Mempool` and `Info`.
 
-```
-git clone https://github.com/tendermint/rust-abci.git
-```
+> Note: Implementations of these traits are expected to be `Send + Sync` and methods take immutable reference of `self`.
+So, internal mutability must be handled using thread safe (`Arc`, `Mutex`, etc.) constructs.
 
-The `empty_app` example, found under the `examples` folder, is a good demonstration/bare minimum foundation for a Rust ABCI app.
+After implementing all three above mentioned `trait`s, you can create a `Server` object and use `Server::run()` to start
+ABCI application.
 
-To use this library to build your own ABCI apps in Rust you have to include the following in your `Cargo.toml` file.
+`Server::run()` is an `async` function and returns a `Future`. So, you'll need an executor to drive `Future` returned
+from `Server::run()`. `async-std` and `tokio` are two popular options. In `counter` example, we use `tokio`'s executor.
 
-```toml
-[dependencies]
-abci = "0.7.0"
-```
+To know more, go to `examples/` to see a sample ABCI application.
+
+### Features
+
+- `tokio`: Enables `tokio` backend for running ABCI TCP/UDS server
+  - **Enabled** by default.
+- `async-std`: Enables `async-std` backend for running ABCI TCP/UDS server
+  - **Disabled** by default.
+
+> Features `tokio` and `async-std` are mutually exclusive, i.e., only one of them can be enabled at a time. Compilation
+will fail if either both of them are enabled or none of them are enabled.
 
 ### Development
 
@@ -50,9 +62,7 @@ To run either of the example apps you have to have Tendermint installed and init
 tendermint node
 ```
 
-After the node is online, you can run the `empty_app` example using `cargo run --example empty_app`.
-
-To run the `counter_app` run `cargo run --example counter_app` and send transaction to Tendermint via:
+After the node is online, you can run the `counter` example using `cargo run --example counter`.
 
 ```
 curl localhost:26657/broadcast_tx_commit?tx=0x01

build.rs

@@ -2,7 +2,7 @@ extern crate protobuf_codegen_pure;
 
 fn main() {
     protobuf_codegen_pure::run(protobuf_codegen_pure::Args {
-        out_dir: "src/messages",
+        out_dir: "src/proto",
         input: &[
             "protobuf/abci.proto",
             "protobuf/libs/kv/types.proto",

examples/counter.rs

@@ -0,0 +1,179 @@
+use std::{
+    net::SocketAddr,
+    sync::{Arc, Mutex},
+};
+
+use abci::{async_trait, types::*, Consensus, Info, Mempool, Server};
+
+/// Simple counter
+#[derive(Debug, Default, Clone)]
+pub struct CounterState {
+    block_height: i64,
+    app_hash: Vec<u8>,
+    counter: u64,
+}
+
+#[derive(Debug)]
+pub struct ConsensusConnection {
+    committed_state: Arc<Mutex<CounterState>>,
+    current_state: Arc<Mutex<Option<CounterState>>>,
+}
+
+impl ConsensusConnection {
+    pub fn new(
+        committed_state: Arc<Mutex<CounterState>>,
+        current_state: Arc<Mutex<Option<CounterState>>>,
+    ) -> Self {
+        Self {
+            committed_state,
+            current_state,
+        }
+    }
+}
+
+#[async_trait]
+impl Consensus for ConsensusConnection {
+    async fn init_chain(&self, _init_chain_request: InitChainRequest) -> InitChainResponse {
+        Default::default()
+    }
+
+    async fn begin_block(&self, _begin_block_request: BeginBlockRequest) -> BeginBlockResponse {
+        let committed_state = self.committed_state.lock().unwrap().clone();
+
+        let mut current_state = self.current_state.lock().unwrap();
+        *current_state = Some(committed_state);
+
+        Default::default()
+    }
+
+    async fn deliver_tx(&self, deliver_tx_request: DeliverTxRequest) -> Result<DeliverTxResponse> {
+        let new_counter = parse_bytes_to_counter(&deliver_tx_request.tx)?;
+
+        let mut current_state_lock = self.current_state.lock().unwrap();
+        let mut current_state = current_state_lock.as_mut().unwrap();
+
+        if current_state.counter + 1 != new_counter {
+            return Err(Error {
+                code: 2,
+                codespace: "Validation error".to_owned(),
+                log: "Only consecutive integers are allowed".to_owned(),
+                info: "Numbers to counter app should be supplied in increasing order of consecutive integers staring from 1".to_owned(),
+            });
+        }
+
+        current_state.counter = new_counter;
+
+        Ok(Default::default())
+    }
+
+    async fn end_block(&self, end_block_request: EndBlockRequest) -> EndBlockResponse {
+        let mut current_state_lock = self.current_state.lock().unwrap();
+        let mut current_state = current_state_lock.as_mut().unwrap();
+
+        current_state.block_height = end_block_request.height;
+        current_state.app_hash = current_state.counter.to_be_bytes().to_vec();
+
+        Default::default()
+    }
+
+    async fn commit(&self) -> CommitResponse {
+        let current_state = self.current_state.lock().unwrap().as_ref().unwrap().clone();
+        let mut committed_state = self.committed_state.lock().unwrap();
+        *committed_state = current_state;
+
+        CommitResponse {
+            data: (*committed_state).app_hash.clone(),
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct MempoolConnection {
+    state: Arc<Mutex<Option<CounterState>>>,
+}
+
+impl MempoolConnection {
+    pub fn new(state: Arc<Mutex<Option<CounterState>>>) -> Self {
+        Self { state }
+    }
+}
+
+#[async_trait]
+impl Mempool for MempoolConnection {
+    async fn check_tx(&self, check_tx_request: CheckTxRequest) -> Result<CheckTxResponse> {
+        let new_counter = parse_bytes_to_counter(&check_tx_request.tx)?;
+
+        let state_lock = self.state.lock().unwrap();
+        let state = state_lock.as_ref().unwrap();
+
+        if state.counter + 1 != new_counter {
+            Err(Error {
+                code: 2,
+                codespace: "Validation error".to_owned(),
+                log: "Only consecutive integers are allowed".to_owned(),
+                info: "Numbers to counter app should be supplied in increasing order of consecutive integers staring from 1".to_owned(),
+            })
+        } else {
+            Ok(Default::default())
+        }
+    }
+}
+
+pub struct InfoConnection {
+    state: Arc<Mutex<CounterState>>,
+}
+
+impl InfoConnection {
+    pub fn new(state: Arc<Mutex<CounterState>>) -> Self {
+        Self { state }
+    }
+}
+
+#[async_trait]
+impl Info for InfoConnection {
+    async fn info(&self, _info_request: InfoRequest) -> InfoResponse {
+        let state = self.state.lock().unwrap();
+
+        InfoResponse {
+            data: Default::default(),
+            version: Default::default(),
+            app_version: Default::default(),
+            last_block_height: (*state).block_height,
+            last_block_app_hash: (*state).app_hash.clone(),
+        }
+    }
+}
+
+fn parse_bytes_to_counter(bytes: &[u8]) -> Result<u64> {
+    if bytes.len() != 8 {
+        return Err(Error {
+            code: 1,
+            codespace: "Parsing error".to_owned(),
+            log: "Transaction should be 8 bytes long".to_owned(),
+            info: "Transaction is big-endian encoding of 64-bit integer".to_owned(),
+        });
+    }
+
+    let mut counter_bytes = [0; 8];
+    counter_bytes.copy_from_slice(bytes);
+
+    Ok(u64::from_be_bytes(counter_bytes))
+}
+
+#[tokio::main]
+async fn main() -> std::io::Result<()> {
+    env_logger::init();
+
+    let committed_state: Arc<Mutex<CounterState>> = Default::default();
+    let current_state: Arc<Mutex<Option<CounterState>>> = Default::default();
+
+    let consensus = ConsensusConnection::new(committed_state.clone(), current_state.clone());
+    let mempool = MempoolConnection::new(current_state.clone());
+    let info = InfoConnection::new(committed_state.clone());
+
+    let server = Server::new(consensus, mempool, info, false);
+
+    server
+        .run("127.0.0.1:26658".parse::<SocketAddr>().unwrap())
+        .await
+}