Cleanup (#254)

* Return Result when creating DaprHttpServer

Especially when running custom docker setups, the container that
contains the sidecar may not be running exactly when the Rust code
starts running.

To fix this, before we needed to sleep(2s) to avoid a panic in the Rust
program.
With this patch, this can be handled on the user side (e.g. the
connection can be retried multiple times with timeouts in-between).

Signed-off-by: Leon Matthes <leon.matthes@kdab.com>

* release: v0.16.0-rc.4

Signed-off-by: Mike Nguyen <hey@mike.ee>

* chore(deps): remove unused crates

Signed-off-by: Mike Nguyen <hey@mike.ee>

* doc: missing expression closure

Signed-off-by: Mike Nguyen <hey@mike.ee>

* refactor: lint issues and correctness improvements

Signed-off-by: Mike Nguyen <hey@mike.ee>

---------

Signed-off-by: Leon Matthes <leon.matthes@kdab.com>
Signed-off-by: Mike Nguyen <hey@mike.ee>
Co-authored-by: Leon Matthes <leon.matthes@kdab.com>

.github/workflows/dapr-bot/Cargo.toml

@@ -11,6 +11,6 @@ rust-version = "1.70.0"
 
 [dependencies]
 exitcode = "1.1.2"
-octocrab = "0.34.1"
+octocrab = "0.42.1"
 serde_json = "1.0.114"
 tokio = { version = "1.36.0", features = ["full"] }

Cargo.toml

@@ -13,7 +13,6 @@ resolver = "2"
 [workspace.dependencies]
 async-trait = "0.1"
 prost = "0.13.1"
-prost-build = "0.13.1"
 prost-types = "0.13.1"
 
 serde = "1.0"
@@ -21,14 +20,13 @@ serde_json = "1.0"
 
 tokio = "1.39"
 tokio-stream = "0.1"
-tokio-test = "0.4"
 tokio-util = "0.7"
 
 tonic = "0.12.1"
 tonic-build = "0.12.1"
 
 [workspace.package]
-version = "0.16.0-rc.3"
+version = "0.16.0-rc.4"
 authors = [
     "Mike Nguyen <hey@mike.ee>",
     "The Dapr Authors <dapr@dapr.io>"

README.md

@@ -53,7 +53,7 @@ Add the following to your `Cargo.toml` file:
 
 ```toml
 [dependencies]
-dapr = "0.16.0-rc.3"
+dapr = "0.16.0-rc.4"
 ```
 
 Here's a basic example to create a client:

dapr-macros/Cargo.toml

@@ -12,9 +12,6 @@ rust-version.workspace = true
 proc-macro = true
 
 [dependencies]
-async-trait = { workspace = true }
-axum = "0.7"
-log = "0.4"
 proc-macro2 = "1.0"
 quote = "1.0"
 syn = { version = "2.0", features = ["full"] }

dapr/Cargo.toml

@@ -13,7 +13,6 @@ rust-version.workspace = true
 [dependencies]
 async-trait = { workspace = true }
 axum = "0.7"
-bytes = "1.7"
 chrono = "0.4"
 futures = "0.3"
 log = "0.4"
@@ -33,5 +32,4 @@ dapr = { path = "./" }
 dapr-macros = { path = "../dapr-macros" }
 tokio = { workspace = true, features = ["full"] }
 uuid = { version = "1.10", features = ["v4"] }
-tokio-test = { workspace = true }
 tokio-stream = { workspace = true }

dapr/src/appcallback.rs

@@ -1,37 +1,36 @@
-use crate::dapr;
 use crate::dapr::proto::runtime::v1::app_callback_server::AppCallback;
 use crate::dapr::proto::{common, runtime};
 use std::collections::HashMap;
 use tonic::{Code, Request, Response, Status};
 
 /// InvokeRequest is the message to invoke a method with the data.
-pub type InvokeRequest = dapr::proto::common::v1::InvokeRequest;
+pub type InvokeRequest = common::v1::InvokeRequest;
 
 /// InvokeResponse is the response message inclduing data and its content type
 /// from app callback.
-pub type InvokeResponse = dapr::proto::common::v1::InvokeResponse;
+pub type InvokeResponse = common::v1::InvokeResponse;
 
 /// ListTopicSubscriptionsResponse is the message including the list of the subscribing topics.
-pub type ListTopicSubscriptionsResponse = dapr::proto::runtime::v1::ListTopicSubscriptionsResponse;
+pub type ListTopicSubscriptionsResponse = runtime::v1::ListTopicSubscriptionsResponse;
 
 /// TopicSubscription represents a topic and it's metadata (session id etc.)
-pub type TopicSubscription = dapr::proto::runtime::v1::TopicSubscription;
+pub type TopicSubscription = runtime::v1::TopicSubscription;
 
 /// TopicEventRequest message is compatiable with CloudEvent spec v1.0.
-pub type TopicEventRequest = dapr::proto::runtime::v1::TopicEventRequest;
+pub type TopicEventRequest = runtime::v1::TopicEventRequest;
 
 /// TopicEventResponse is response from app on published message
-pub type TopicEventResponse = dapr::proto::runtime::v1::TopicEventResponse;
+pub type TopicEventResponse = runtime::v1::TopicEventResponse;
 
 /// ListInputBindingsResponse is the message including the list of input bindings.
-pub type ListInputBindingsResponse = dapr::proto::runtime::v1::ListInputBindingsResponse;
+pub type ListInputBindingsResponse = runtime::v1::ListInputBindingsResponse;
 
 /// BindingEventRequest represents input bindings event.
-pub type BindingEventRequest = dapr::proto::runtime::v1::BindingEventRequest;
+pub type BindingEventRequest = runtime::v1::BindingEventRequest;
 
 /// BindingEventResponse includes operations to save state or
 /// send data to output bindings optionally.
-pub type BindingEventResponse = dapr::proto::runtime::v1::BindingEventResponse;
+pub type BindingEventResponse = runtime::v1::BindingEventResponse;
 
 impl ListTopicSubscriptionsResponse {
     /// Create `ListTopicSubscriptionsResponse` with a topic.

dapr/src/client.rs

@@ -656,6 +656,13 @@ impl DaprInterface for dapr_v1::dapr_client::DaprClient<TonicChannel> {
         Ok(dapr_v1::dapr_client::DaprClient::connect(addr).await?)
     }
 
+    async fn publish_event(&mut self, request: PublishEventRequest) -> Result<(), Error> {
+        self.publish_event(Request::new(request))
+            .await?
+            .into_inner();
+        Ok(())
+    }
+
     async fn invoke_service(
         &mut self,
         request: InvokeServiceRequest,
@@ -676,13 +683,6 @@ impl DaprInterface for dapr_v1::dapr_client::DaprClient<TonicChannel> {
             .into_inner())
     }
 
-    async fn publish_event(&mut self, request: PublishEventRequest) -> Result<(), Error> {
-        self.publish_event(Request::new(request))
-            .await?
-            .into_inner();
-        Ok(())
-    }
-
     async fn get_secret(&mut self, request: GetSecretRequest) -> Result<GetSecretResponse, Error> {
         Ok(self.get_secret(Request::new(request)).await?.into_inner())
     }
@@ -701,6 +701,11 @@ impl DaprInterface for dapr_v1::dapr_client::DaprClient<TonicChannel> {
         Ok(self.get_state(Request::new(request)).await?.into_inner())
     }
 
+    async fn save_state(&mut self, request: SaveStateRequest) -> Result<(), Error> {
+        self.save_state(Request::new(request)).await?.into_inner();
+        Ok(())
+    }
+
     async fn query_state_alpha1(
         &mut self,
         request: QueryStateRequest,
@@ -711,11 +716,6 @@ impl DaprInterface for dapr_v1::dapr_client::DaprClient<TonicChannel> {
             .into_inner())
     }
 
-    async fn save_state(&mut self, request: SaveStateRequest) -> Result<(), Error> {
-        self.save_state(Request::new(request)).await?.into_inner();
-        Ok(())
-    }
-
     async fn delete_state(&mut self, request: DeleteStateRequest) -> Result<(), Error> {
         self.delete_state(Request::new(request)).await?.into_inner();
         Ok(())

daprdocs/content/en/rust-sdk-docs/rust-client/_index.md

@@ -34,7 +34,7 @@ dapr = "0.16.0"
 You can either reference `dapr::Client` or bind the full path to a new name as follows:
 
 ```rust
-use dapr::Client as DaprClient
+use dapr::Client as DaprClient;
 ```
 
 ## Instantiating the Dapr client
@@ -43,7 +43,7 @@ use dapr::Client as DaprClient
 let addr = "https://127.0.0.1".to_string();
 
 let mut client = dapr::Client::<dapr::client::TonicClient>::connect(addr,
-    port).await?;
+port).await?;
 ```
 
 ## Building blocks

examples/Cargo.toml

@@ -10,7 +10,6 @@ rust-version.workspace = true
 
 [dependencies]
 async-trait = { workspace = true }
-base64 = "0.22"
 dapr = { path = "../dapr" }
 dapr-macros = { path = "../dapr-macros" }
 env_logger = "0.11"

examples/src/actors/client.rs

@@ -14,7 +14,7 @@ pub struct MyRequest {
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    std::thread::sleep(std::time::Duration::new(2, 0));
+    tokio::time::sleep(std::time::Duration::new(2, 0)).await;
 
     // Define the Dapr address
     let addr = "https://127.0.0.1".to_string();

examples/src/bindings/input.rs

@@ -61,7 +61,7 @@ impl AppCallback for AppCallbackService {
         let name = &r.name;
         let data = &r.data;
 
-        let message = String::from_utf8_lossy(&data);
+        let message = String::from_utf8_lossy(data);
         println!("Binding Name: {}", &name);
         println!("Message: {}", &message);
 

examples/src/bindings/output.rs

@@ -1,10 +1,10 @@
-use std::{collections::HashMap, thread, time::Duration};
+use std::{collections::HashMap, time::Duration};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    thread::sleep(Duration::from_secs(2));
+    tokio::time::sleep(Duration::from_secs(2)).await;
 
     // Get the Dapr port and create a connection
     let addr = "https://127.0.0.1".to_string();

examples/src/client/client.rs

@@ -2,7 +2,7 @@
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    std::thread::sleep(std::time::Duration::new(2, 0));
+    tokio::time::sleep(std::time::Duration::new(2, 0)).await;
 
     // Set the Dapr address
     let addr = "https://127.0.0.1".to_string();

examples/src/configuration/main.rs

@@ -7,7 +7,7 @@ type DaprClient = dapr::Client<dapr::client::TonicClient>;
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    std::thread::sleep(std::time::Duration::new(2, 0));
+    tokio::time::sleep(std::time::Duration::new(2, 0)).await;
 
     // Set the Dapr address
     let addr = "https://127.0.0.1".to_string();
@@ -19,14 +19,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // get key-value pair in the state store
     let response = client
-        .get_configuration(CONFIGSTORE_NAME, vec![(&key)], None)
+        .get_configuration(CONFIGSTORE_NAME, vec![&key], None)
         .await?;
     let val = response.items.get("hello").unwrap();
     println!("Configuration value: {val:?}");
 
     // Subscribe for configuration changes
     let mut stream = client
-        .subscribe_configuration(CONFIGSTORE_NAME, vec![(&key)], None)
+        .subscribe_configuration(CONFIGSTORE_NAME, vec![&key], None)
         .await?;
 
     let mut subscription_id = String::new();

examples/src/conversation/main.rs

@@ -1,5 +1,4 @@
 use dapr::client::{ConversationInputBuilder, ConversationRequestBuilder};
-use std::thread;
 use std::time::Duration;
 
 type DaprClient = dapr::Client<dapr::client::TonicClient>;
@@ -7,7 +6,7 @@ type DaprClient = dapr::Client<dapr::client::TonicClient>;
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Sleep to allow for the server to become available
-    thread::sleep(Duration::from_secs(5));
+    tokio::time::sleep(Duration::from_secs(5)).await;
 
     // Set the Dapr address
     let address = "https://127.0.0.1".to_string();

examples/src/invoke/grpc-proxying/client.rs

@@ -1,4 +1,4 @@
-use std::{thread, time::Duration};
+use std::time::Duration;
 
 use hello_world::{greeter_client::GreeterClient, HelloRequest};
 
@@ -11,7 +11,7 @@ pub mod hello_world {
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Sleep to allow for the server to become available
-    thread::sleep(Duration::from_secs(5));
+    tokio::time::sleep(Duration::from_secs(5)).await;
 
     // Get the Dapr port and create a connection
     let port: u16 = std::env::var("DAPR_GRPC_PORT").unwrap().parse().unwrap();

examples/src/invoke/grpc/client.rs

@@ -1,5 +1,5 @@
 use crate::hello_world::HelloReply;
-use std::{thread, time::Duration};
+use std::time::Duration;
 
 use prost::Message;
 
@@ -12,7 +12,7 @@ type DaprClient = dapr::Client<dapr::client::TonicClient>;
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Sleep to allow for the server to become available
-    thread::sleep(Duration::from_secs(5));
+    tokio::time::sleep(Duration::from_secs(5)).await;
 
     // Set the Dapr address
     let address = "https://127.0.0.1".to_string();

examples/src/pubsub/publisher.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashMap, thread, time::Duration};
+use std::{collections::HashMap, time::Duration};
 
 use tokio::time;
 
@@ -21,7 +21,7 @@ struct Refund {
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    thread::sleep(Duration::from_secs(2));
+    tokio::time::sleep(Duration::from_secs(2)).await;
 
     // Set address for Dapr connection
     let addr = "https://127.0.0.1".to_string();

examples/src/query_state/query1.rs

@@ -3,7 +3,7 @@ use serde_json::json;
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    std::thread::sleep(std::time::Duration::new(5, 0));
+    tokio::time::sleep(std::time::Duration::new(5, 0)).await;
 
     // Set the Dapr address and create a connection
     let addr = "https://127.0.0.1".to_string();

examples/src/query_state/query2.rs

@@ -3,7 +3,7 @@ use serde_json::json;
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Introduce delay so that dapr grpc port is assigned before app tries to connect
-    std::thread::sleep(std::time::Duration::new(5, 0));
+    tokio::time::sleep(std::time::Duration::new(5, 0)).await;
 
     // Set the Dapr address and create a connection
     let addr = "https://127.0.0.1".to_string();