New API (#7)

* feat(request-response object): added Request and Response object similar to tonic, added expanded-request to test how they should work Signed-off-by: gabrik <[email protected]> * feat(request-response-object): using result in a way similar to tonic, added into example Signed-off-by: gabrik <[email protected]> * feat(typed session): WIP, basic pub/sub working [no ci] Signed-off-by: gabrik <[email protected]> * feat(typed-session): WIP [no ci] Signed-off-by: gabrik <[email protected]> * feat(typed-primitives): bump to latest zenoh development version Signed-off-by: gabrik <[email protected]> * feat(simplified-api): adding simplified API, similar to tower+tonic [no ci] Signed-off-by: gabrik <[email protected]> * feat(liveliness): using liveliness tokens to discover servers [no ci] * feat(new-api): completely redesign of API to match tonic services WIP [no ci] Signed-off-by: gabrik <[email protected]> * feat(new-api): using async-trait instead of custom BoxFuture Signed-off-by: gabrik <[email protected]> * feat(new-api) using zenoh 0.10.1-rc Signed-off-by: gabrik <[email protected]> * chore(ci): update CI policies Signed-off-by: gabrik <[email protected]> * chore: fix format Signed-off-by: gabrik <[email protected]> * chore: update dependencies syn and darling Signed-off-by: gabrik <[email protected]> * chore: format Signed-off-by: gabrik <[email protected]> * chore: better error handling in client, QoL improvements in Status Signed-off-by: gabrik <[email protected]> * chore: QoL improvments in requests and responses Signed-off-by: gabrik <[email protected]> * feat(builder): added builders for client and server, added labels when finding servers Signed-off-by: gabrik <[email protected]> * feat(QoL): some QoL improvments in Status Signed-off-by: gabrik <[email protected]> * feat(tokio): migrating from async_std to tokio Signed-off-by: gabrik <[email protected]> * fix: using fully qualified type path Signed-off-by: gabrik <[email protected]> * fix: adding use inside generated code Signed-off-by: gabrik <[email protected]> * chore: bump zenoh to 0.11.0-rc.2 Signed-off-by: gabrik <[email protected]> * refactor: removing unwraps Signed-off-by: gabrik <[email protected]> * fix: adding missing import that will cause issue in generated code Signed-off-by: gabrik <[email protected]> * chore: using stable rust Signed-off-by: gabrik <[email protected]> * chore(deps): bump zenoh version Signed-off-by: Gabriele Baldoni <[email protected]> * chore(lint): fixing format Signed-off-by: Gabriele Baldoni <[email protected]> --------- Signed-off-by: gabrik <[email protected]> Signed-off-by: Gabriele Baldoni <[email protected]>
ZettaScaleLabs · Jun 20, 2024 · 23a5d17 · 23a5d17
1 parent 74116f8
commit 23a5d17
Show file tree

Hide file tree

Showing 36 changed files with 3,290 additions and 3,286 deletions.
diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml
@@ -15,9 +15,9 @@ name: Rust
 
 on:
   push:
-    branches: [ master ]
+    branches: [ main ]
   pull_request:
-    branches: [ master ]
+    branches: [ "**" ]
 
 jobs:
   build:

diff --git a/Cargo.toml b/Cargo.toml
@@ -12,18 +12,13 @@
 #   ADLINK fog05 team, <[email protected]>
 #
 [workspace]
+resolver = "2"
 
-members = [
-  "zrpc",
-  "zrpc-macros",
-]
+members = ["zrpc", "zrpc-derive"]
 
-exclude = [
-  "zrpc-perf",
-]
+exclude = ["zenoh-typed", "zrpc-perf"]
 
 [workspace.package]
-version = "0.7.0-alpha.1"
 authors = ["gabrik <[email protected]>"]
 categories = ["network-programming"]
 description = "Framework for decentralized and distributed microservices over Zenoh."
@@ -32,11 +27,45 @@ homepage = "https://github.com/ZettaScaleLabs/zenoh-rpc"
 license = " EPL-2.0 OR Apache-2.0"
 readme = "README.md"
 repository = "https://github.com/ZettaScaleLabs/zenoh-rpc"
-
+version = "0.8.0-alpha.1"
 
 [profile.release]
-debug=false
-lto="fat"
-codegen-units=1
-opt-level=3
-panic="abort"
+codegen-units = 1
+debug = false
+lto = "fat"
+opt-level = 3
+panic = "abort"
+
+[workspace.dependencies]
+Inflector = "0.11"
+async-trait = "0.1.79"
+base64 = "0.22.0"
+bincode = { version = "1.3.3" }
+clap = { version = "4.4.11", features = ["derive"] }
+darling = "0.20"
+env_logger = "0.11.0"
+flume = { version = "0.11" }
+futures = "0.3.28"
+log = "0.4"
+proc-macro2 = "1.0"
+quote = "1.0"
+serde = { version = "1.0.160", features = ["derive"] }
+serde_cbor = { version = "0.11.2" }
+serde_derive = "1.0.160"
+serde_json = { version = "1.0.114" }
+serde_yaml = { version = "0.9" }
+syn = { version = "2.0", features = ["full", "visit-mut"] }
+syn-serde = { version = "0.3", features = ["json"] }
+thiserror = "1.0"
+tokio = { version = "1.35.1", default-features = false, features = [
+  "macros",
+  "rt",
+  "time",
+] }
+zenoh = { version = "0.11.0", default-features = false }
+zenoh-codec = { version = "0.11.0" }
+zenoh-core = { version = "0.11.0" }
+zenoh-ext = { version = "0.11.0" }
+zenoh-macros = { version = "0.11.0" }
+zenoh-protocol = { version = "0.11.0" }
+zenoh-util = { version = "0.11.0" }
diff --git a/rust-toolchain b/rust-toolchain
@@ -1 +1 @@
-1.69.0
+stable
diff --git a/zenoh-typed/Cargo.toml b/zenoh-typed/Cargo.toml
@@ -0,0 +1,52 @@
+[package]
+authors.workspace = true
+categories.workspace = true
+description.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+name = "zenoh-typed"
+readme.workspace = true
+repository.workspace = true
+version.workspace = true
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+base64 = { workspace = true }
+bincode = { workspace = true, optional = true }
+flume = { workspace = true }
+futures = { workspace = true }
+log = { workspace = true }
+serde = { workspace = true }
+serde_cbor = { workspace = true }
+serde_derive = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["io-std","io-util", "rt-multi-thread"]}
+zenoh = { workspace = true }
+zenoh-codec = { workspace = true }
+zenoh-core = { workspace = true }
+zenoh-macros = { workspace = true }
+zenoh-protocol = { workspace = true }
+zenoh-util = { workspace = true }
+
+[dev-dependencies]
+clap = { workspace = true }
+env_logger = { workspace = true }
+
+[features]
+#zenoh feature re-exports
+auth_pubkey = ["zenoh/auth_pubkey"]
+auth_usrpwd = ["zenoh/auth_usrpwd"]
+shared-memory = ["zenoh/shared-memory"]
+stats = ["zenoh/stats"]
+transport_quic = ["zenoh/transport_quic"]
+transport_tcp = ["zenoh/transport_tcp"]
+transport_tls = ["zenoh/transport_tls"]
+transport_udp = ["zenoh/transport_udp"]
+transport_unixsock-stream = ["zenoh/transport_unixsock-stream"]
+transport_ws = ["zenoh/transport_ws"]
+
+query_payload = ["zenoh/unstable"]
+
+default = ["query_payload", "transport_tcp", "transport_udp"]
diff --git a/zenoh-typed/examples/z_pub.rs b/zenoh-typed/examples/z_pub.rs
@@ -0,0 +1,157 @@
+//
+// Copyright (c) 2023 ZettaScale Technology
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License 2.0 which is available at
+// http://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+//
+// Contributors:
+//   ZettaScale Zenoh Team, <[email protected]>
+//
+use tokio::time::sleep;
+use clap::Parser;
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+use zenoh::config::Config;
+// use zenoh::payload::Serialize;
+use zenoh::prelude::r#async::*;
+use zenoh_typed::prelude::*;
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+struct MyData {
+    pub name: String,
+    pub id: u64,
+}
+
+#[tokio::main]
+async fn main() {
+    // Initiate logging
+    env_logger::init();
+
+    let (config, key_expr, value, _attachment) = parse_args();
+
+    println!("Opening session...");
+    let session = zenoh::open(config).res().await.unwrap();
+    let session = SerdeSession::new(session, CBOR);
+
+    println!("Declaring Publisher on '{key_expr}'...");
+    let publisher = session
+        .declare_publisher::<_, MyData>(key_expr)
+        .res()
+        .await
+        .unwrap();
+
+    for idx in 0..u32::MAX {
+        sleep(Duration::from_secs(1)).await;
+        let value: MyData = MyData {
+            name: value.clone(),
+            id: idx as u64,
+        };
+        println!("Putting Data ('{value:?}')...");
+
+        publisher.put(value).res().await.unwrap();
+
+        // TypedSession::<CBOREncoder, CBOREncoder>::put(&session, &key_expr, &value)
+        //     .res()
+        //     .await
+        //     .unwrap();
+    }
+}
+
+#[derive(Parser, Clone, PartialEq, Eq, Hash, Debug)]
+struct Args {
+    #[arg(short, long, default_value = "demo/example/zenoh-rs-pub")]
+    /// The key expression to write to.
+    key: KeyExpr<'static>,
+    #[arg(short, long, default_value = "Pub from Rust!")]
+    /// The value to write.
+    value: String,
+    #[arg(short, long)]
+    /// The attachments to add to each put.
+    ///
+    /// The key-value pairs are &-separated, and = serves as the separator between key and value.
+    attach: Option<String>,
+    #[command(flatten)]
+    common: CommonArgs,
+}
+
+fn parse_args() -> (Config, KeyExpr<'static>, String, Option<String>) {
+    let args = Args::parse();
+    (args.common.into(), args.key, args.value, args.attach)
+}
+
+#[derive(clap::ValueEnum, Clone, Copy, PartialEq, Eq, Hash, Debug)]
+pub enum Wai {
+    Peer,
+    Client,
+    Router,
+}
+impl core::fmt::Display for Wai {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        core::fmt::Debug::fmt(&self, f)
+    }
+}
+#[derive(clap::Parser, Clone, PartialEq, Eq, Hash, Debug)]
+pub struct CommonArgs {
+    #[arg(short, long)]
+    /// A configuration file.
+    config: Option<String>,
+    #[arg(short, long)]
+    /// The Zenoh session mode [default: peer].
+    mode: Option<Wai>,
+    #[arg(short = 'e', long)]
+    /// Endpoints to connect to.
+    connect: Vec<String>,
+    #[arg(short, long)]
+    /// Endpoints to listen on.
+    listen: Vec<String>,
+    #[arg(long)]
+    /// Disable the multicast-based scouting mechanism.
+    no_multicast_scouting: bool,
+    #[arg(long)]
+    /// Disable the multicast-based scouting mechanism.
+    enable_shm: bool,
+}
+
+impl From<CommonArgs> for Config {
+    fn from(value: CommonArgs) -> Self {
+        (&value).into()
+    }
+}
+impl From<&CommonArgs> for Config {
+    fn from(value: &CommonArgs) -> Self {
+        let mut config = match &value.config {
+            Some(path) => Config::from_file(path).unwrap(),
+            None => Config::default(),
+        };
+        match value.mode {
+            Some(Wai::Peer) => config.set_mode(Some(zenoh::scouting::WhatAmI::Peer)),
+            Some(Wai::Client) => config.set_mode(Some(zenoh::scouting::WhatAmI::Client)),
+            Some(Wai::Router) => config.set_mode(Some(zenoh::scouting::WhatAmI::Router)),
+            None => Ok(None),
+        }
+        .unwrap();
+        if !value.connect.is_empty() {
+            config.connect.endpoints = value.connect.iter().map(|v| v.parse().unwrap()).collect();
+        }
+        if !value.listen.is_empty() {
+            config.listen.endpoints = value.listen.iter().map(|v| v.parse().unwrap()).collect();
+        }
+        if value.no_multicast_scouting {
+            config.scouting.multicast.set_enabled(Some(false)).unwrap();
+        }
+        if value.enable_shm {
+            #[cfg(feature = "shared-memory")]
+            config.transport.shared_memory.set_enabled(true).unwrap();
+            #[cfg(not(feature = "shared-memory"))]
+            {
+                println!("enable-shm argument: SHM cannot be enabled, because Zenoh is compiled without shared-memory feature!");
+                std::process::exit(-1);
+            }
+        }
+        config
+    }
+}