Merge pull request #21 from EspressoSystems/api_test

Broadcast channel based event streaming api

.github/actions/install-capnp/action.yml

@@ -0,0 +1,8 @@
+name: Install Capn Proto
+runs:
+  using: composite
+  steps:
+    - run: |
+        sudo apt-get update
+        sudo apt-get install -y capnproto
+      shell: bash

.github/workflows/build.yml

@@ -29,6 +29,9 @@ jobs:
 
       - name: Install Protoc
         uses: arduino/setup-protoc@v2
+	
+      - name: Install capnproto
+         uses: ./.github/actions/install-capnp
 
       - uses: dtolnay/rust-toolchain@stable
 
@@ -51,8 +54,9 @@ jobs:
           token: ${{ github.token }}
           args: --workspace --all-features --all-targets -- -D warnings
 
-      - name: Audit
-        run: cargo audit --ignore RUSTSEC-2023-0018 --ignore RUSTSEC-2023-0052 --ignore RUSTSEC-2023-0065
+      # TODO: This should not be the basis of a build failure. Move to a different job
+       # - name: Audit
+       #   run: cargo audit --ignore RUSTSEC-2023-0018 --ignore RUSTSEC-2023-0052 --ignore RUSTSEC-2023-0065
 
       - name: Build
         # Build in release without `testing` feature, this should work without `hotshot_example` config.

.github/workflows/debug_build.yml

@@ -15,6 +15,9 @@ jobs:
       - name: Install Protoc
         uses: arduino/setup-protoc@v2
 
+      - name: Install capnproto
+        uses: ./.github/actions/install-capnp
+
       - uses: dtolnay/rust-toolchain@stable
 
       - uses: styfle/[email protected]

.github/workflows/lint.yml

@@ -36,6 +36,9 @@ jobs:
       - name: Install Protoc
         uses: arduino/setup-protoc@v2
 
+      - name: Install capnproto
+        uses: ./.github/actions/install-capnp
+
       - uses: dtolnay/rust-toolchain@stable
 
       - name: Configure Git

Cargo.toml

@@ -6,14 +6,28 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-hotshot-query-service = { git = "https://github.com/EspressoSystems/hotshot-query-service.git", tag = "0.0.15" }
-async-trait = "0.1"
+async-broadcast = "0.7.0"
 clap = { version = "4.4", features = ["derive", "env"] }
+async-std = { version = "1.9.0", features = ["unstable", "attributes"] }
+async-compatibility-layer = { git = "https://github.com/EspressoSystems/async-compatibility-layer.git", tag = "1.4.2", features = [
+    "logging-utils",
+] }
+async-lock = "2.8"
+async-trait = "0.1"
 derive_more = "0.99"
+derivative = "2.2"
+either = { version = "1.10", features = ["serde"] }
 futures = "0.3"
-hotshot-types = { git = "https://github.com/EspressoSystems/hotshot-types", tag = "0.1.5" }
+hotshot-types = { git = "https://github.com/EspressoSystems/HotShot.git", tag = "0.5.28" }
 serde = { version = "1.0", features = ["derive"] }
-snafu = { version = "0.7", features = ["backtraces"] }
+snafu = "0.8"
 tagged-base64 = { git = "https://github.com/EspressoSystems/tagged-base64", tag = "0.3.4" }
-tide-disco = { git = "https://github.com/EspressoSystems/tide-disco.git", tag = "v0.4.6" }
+tide-disco = { git = "https://github.com/EspressoSystems/tide-disco.git", tag = "v0.5.0" }
 toml = "0.8"
+versioned-binary-serialization = { git = "https://github.com/EspressoSystems/versioned-binary-serialization", tag = "0.1.2" }
+tracing = "0.1"
+
+[dev-dependencies]
+hotshot-example-types = { git = "https://github.com/EspressoSystems/HotShot.git", tag = "0.5.28" }
+portpicker = "0.1.1"
+surf-disco = { git = "https://github.com/EspressoSystems/surf-disco.git", tag = "v0.5.0" }

api/hotshot_events.toml

@@ -26,9 +26,9 @@ NAME = "hotshot_events_service"
 DESCRIPTION = ""
 FORMAT_VERSION = "0.1.0"
 
-[route.hotshot_events]
-PATH = ["hotshotevents/:view_number"]
-":view_number" = "Integer"
+[route.events]
+PATH = ["events"]
+METHOD = "SOCKET"
 DOC = """
-Get hotshot events starting from a given view number.
+Get hotshot events starting now.
 """

src/api.rs

@@ -14,12 +14,13 @@ use std::fs;
 use std::path::Path;
 use tide_disco::api::{Api, ApiError};
 use toml::{map::Entry, Value};
+use versioned_binary_serialization::version::StaticVersionType;
 
-pub(crate) fn load_api<State, Error>(
+pub(crate) fn load_api<State, Error, Ver: StaticVersionType>(
     path: Option<impl AsRef<Path>>,
     default: &str,
     extensions: impl IntoIterator<Item = Value>,
-) -> Result<Api<State, Error>, ApiError> {
+) -> Result<Api<State, Error, Ver>, ApiError> {
     let mut toml = match path {
         Some(path) => load_toml(path.as_ref())?,
         None => toml::from_str(default).map_err(|err| ApiError::CannotReadToml {

src/events.rs

@@ -1,16 +1,16 @@
 use clap::Args;
 use derive_more::From;
 use futures::{FutureExt, StreamExt, TryFutureExt};
-use hotshot_types::traits::{node_implementation::NodeType, signature_key::SignatureKey};
+use hotshot_types::traits::node_implementation::NodeType;
 use serde::{Deserialize, Serialize};
 use snafu::Snafu;
-use std::{fmt::Display, path::PathBuf};
-use tagged_base64::TaggedBase64;
+use std::path::PathBuf;
 use tide_disco::{api::ApiError, method::ReadState, Api, RequestError, StatusCode};
+use versioned_binary_serialization::version::StaticVersionType;
 
 use crate::{api::load_api, events_source::EventsSource};
 
-#[derive(Args, Default)]
+#[derive(Args, Default, Debug)]
 pub struct Options {
     #[arg(
         long = "hotshot-events-service-api-path",
@@ -80,37 +80,26 @@ impl tide_disco::error::Error for Error {
     }
 }
 
-pub fn define_api<State, Types: NodeType>(options: &Options) -> Result<Api<State, Error>, ApiError>
+pub fn define_api<State, Types: NodeType, Ver: StaticVersionType + 'static>(
+    options: &Options,
+) -> Result<Api<State, Error, Ver>, ApiError>
 where
     State: 'static + Send + Sync + ReadState,
     <State as ReadState>::State: Send + Sync + EventsSource<Types>,
     Types: NodeType,
-    <<Types as NodeType>::SignatureKey as SignatureKey>::PureAssembledSignatureType:
-        for<'a> TryFrom<&'a TaggedBase64> + Into<TaggedBase64> + Display,
-    for<'a> <<<Types as NodeType>::SignatureKey as SignatureKey>::PureAssembledSignatureType as TryFrom<
-        &'a TaggedBase64,
-    >>::Error: Display,
 {
-    let mut api = load_api::<State, Error>(
+    let mut api = load_api::<State, Error, Ver>(
         options.api_path.as_ref(),
         include_str!("../api/hotshot_events.toml"),
         options.extensions.clone(),
     )?;
     api.with_version("0.1.0".parse().unwrap())
-        .stream("hotshot_events", move |req, state| {
+        .stream("events", move |_, state| {
             async move {
-                let view_number = req.integer_param("view_number")?;
-                Ok(state
-                    .read(|state| {
-                        async move {
-                            state
-                                .get_available_hotshot_events(view_number)
-                                .await
-                                .map(Ok)
-                        }
-                        .boxed()
-                    })
-                    .await)
+                tracing::info!("client subscribed to events");
+                state
+                    .read(|state| async move { Ok(state.subscribe_events().await.map(Ok)) }.boxed())
+                    .await
             }
             .try_flatten_stream()
             .boxed()