Merge pull request dapr#87 from zedgell/feature/pubsub-macro

Created Procedural Macros For Pub Sub

Author: mikeee

examples/pubsub/README.md

@@ -3,7 +3,7 @@
 This is a simple example that demonstrates Dapr's pub/sub capabilities. To implement pub/sub in your rust application, you need to implement `AppCallback` server for subscribing to events. Specifically, the following two methods need to be implemented for pub/sub to work:
 
 1. `list_topic_subscriptions` - Dapr runtime calls this method to get list of topics the application is subscribed to.
-2. `on_topic_event` - Defines how the application handles the topic event. 
+2. `on_topic_event` - Defines how the application handles the topic event.
 
 > **Note:** Make sure to use latest version of proto bindings.
 
@@ -54,4 +54,4 @@ dapr run --app-id rust-subscriber --app-protocol grpc --app-port 50051 cargo run
 2. Run the publisher with dapr
 ```bash
 dapr run --app-id rust-publisher --app-protocol grpc cargo run -- --example publisher
-```
+```

examples/pubsub/publisher.rs

@@ -1,5 +1,20 @@
 use std::{collections::HashMap, thread, time::Duration};
 
+use dapr::serde::{Deserialize, Serialize};
+use dapr::serde_json;
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Order {
+    pub order_number: i32,
+    pub order_details: String,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+struct Refund {
+    pub order_number: i32,
+    pub refund_amount: i32,
+}
+
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // TODO: Handle this issue in the sdk
@@ -21,14 +36,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     // topic to publish message to
     let topic = "A".to_string();
+    let topic_b = "B".to_string();
 
-    for count in 0..3 {
+    for count in 0..10 {
+        let order = Order {
+            order_number: count,
+            order_details: format!("Count is {}", count),
+        };
         // message metadata
         let mut metadata = HashMap::<String, String>::new();
         metadata.insert("count".to_string(), count.to_string());
 
         // message
-        let message = format!("{} => hello from rust!", &count).into_bytes();
+        let message = serde_json::to_string(&order).unwrap().into_bytes();
 
         client
             .publish_event(
@@ -43,7 +63,31 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         // sleep for 1 second to simulate delay between each event
         tokio::time::sleep(Duration::from_secs(1)).await;
     }
+    for count in 0..10 {
+        let refund = Refund {
+            order_number: count,
+            refund_amount: 1200,
+        };
+        // message metadata
+        let mut metadata = HashMap::<String, String>::new();
+        metadata.insert("count".to_string(), count.to_string());
+
+        // message
+        let message = serde_json::to_string(&refund).unwrap().into_bytes();
 
+        client
+            .publish_event(
+                &pubsub_name,
+                &topic_b,
+                &data_content_type,
+                message,
+                Some(metadata),
+            )
+            .await?;
+
+        // sleep for 2 seconds to simulate delay between two events
+        tokio::time::sleep(Duration::from_secs(2)).await;
+    }
     println!("messages published");
 
     Ok(())

examples/pubsub/subscriber.rs

@@ -1,78 +1,43 @@
-use tonic::{transport::Server, Request, Response, Status};
+use dapr_macros::topic;
+use tonic::transport::Server;
 
+use dapr::appcallback::AppCallbackService;
+use dapr::serde::{Deserialize, Serialize};
 use dapr::{
-    appcallback::*,
-    dapr::dapr::proto::runtime::v1::app_callback_server::{AppCallback, AppCallbackServer},
+    appcallback::*, dapr::dapr::proto::runtime::v1::app_callback_server::AppCallbackServer,
 };
 
-#[derive(Default)]
-pub struct AppCallbackService {}
-
-#[tonic::async_trait]
-impl AppCallback for AppCallbackService {
-    /// Invokes service method with InvokeRequest.
-    async fn on_invoke(
-        &self,
-        _request: Request<InvokeRequest>,
-    ) -> Result<Response<InvokeResponse>, Status> {
-        Ok(Response::new(InvokeResponse::default()))
-    }
-
-    /// Lists all topics subscribed by this app.
-    ///
-    /// NOTE: Dapr runtime will call this method to get
-    /// the list of topics the app wants to subscribe to.
-    /// In this example, the app is subscribing to topic `A`.
-    async fn list_topic_subscriptions(
-        &self,
-        _request: Request<()>,
-    ) -> Result<Response<ListTopicSubscriptionsResponse>, Status> {
-        let topic = "A".to_string();
-        let pubsub_name = "pubsub".to_string();
-
-        let list_subscriptions = ListTopicSubscriptionsResponse::topic(pubsub_name, topic);
-
-        Ok(Response::new(list_subscriptions))
-    }
-
-    /// Subscribes events from Pubsub.
-    async fn on_topic_event(
-        &self,
-        request: Request<TopicEventRequest>,
-    ) -> Result<Response<TopicEventResponse>, Status> {
-        let r = request.into_inner();
-        let data = &r.data;
-        let data_content_type = &r.data_content_type;
-
-        let message = String::from_utf8_lossy(data);
-        println!("Message: {}", &message);
-        println!("Content-Type: {}", &data_content_type);
+#[derive(Serialize, Deserialize, Debug)]
+struct Order {
+    pub order_number: i32,
+    pub order_details: String,
+}
 
-        Ok(Response::new(TopicEventResponse::default()))
-    }
+#[derive(Serialize, Deserialize, Debug)]
+struct Refund {
+    pub order_number: i32,
+    pub refund_amount: i32,
+}
 
-    /// Lists all input bindings subscribed by this app.
-    async fn list_input_bindings(
-        &self,
-        _request: Request<()>,
-    ) -> Result<Response<ListInputBindingsResponse>, Status> {
-        Ok(Response::new(ListInputBindingsResponse::default()))
-    }
+#[topic(pub_sub_name = "pubsub", topic = "A")]
+async fn handle_a_event(order: Order) {
+    println!("Topic A - {:#?}", order)
+}
 
-    /// Listens events from the input bindings.
-    async fn on_binding_event(
-        &self,
-        _request: Request<BindingEventRequest>,
-    ) -> Result<Response<BindingEventResponse>, Status> {
-        Ok(Response::new(BindingEventResponse::default()))
-    }
+#[topic(pub_sub_name = "pubsub", topic = "B")]
+async fn handle_b_event(refund: Refund) {
+    println!("Topic B - {:#?}", refund)
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let addr = "[::]:50051".parse().unwrap();
+    let addr = "127.0.0.1:50051".parse().unwrap();
+
+    let mut callback_service = AppCallbackService::new();
+
+    callback_service.add_handler(HandleAEvent::default().get_handler());
 
-    let callback_service = AppCallbackService::default();
+    callback_service.add_handler(HandleBEvent::default().get_handler());
 
     println!("AppCallback server listening on: {}", addr);
 

macros/Cargo.toml

@@ -12,5 +12,6 @@ proc-macro = true
 async-trait = "0.1"
 log = "0.4"
 axum = "0.7.4"
-syn = {version="2.0.29",features=["full"]}
+syn = { version = "2.0.29", features = ["full"] }
 quote = "1.0.8"
+proc-macro2 = "1.0"