[ISSUE apache#4519]Rust sdk support CloudEvents (apache#4520)

* [ISSUE apache#4519]Rust sdk support CloudEvents * optimize code * fix complie error
pmupkin · Nov 2, 2023 · 40a2aea · 40a2aea
1 parent 4bda079
commit 40a2aea
Show file tree

Hide file tree

Showing 16 changed files with 400 additions and 117 deletions.
diff --git a/eventmesh-sdks/eventmesh-sdk-rust/Cargo.toml b/eventmesh-sdks/eventmesh-sdk-rust/Cargo.toml
@@ -26,15 +26,16 @@ description = "Rust client for Apache EventMesh"
 license = "Apache-2.0"
 keywords = ["EventMesh", "SDK", "rust-client", "rust", "eventmesh-rust-sdk"]
 readme = "./README.md"
-homepage      = "https://github.com/apache/eventmesh"
-repository    = "https://github.com/apache/eventmesh"
+homepage = "https://github.com/apache/eventmesh"
+repository = "https://github.com/apache/eventmesh"
 
 [features]
-default = ["grpc","eventmesh_message"]
-full = ["grpc","eventmesh_message"]
-eventmesh_message=[]
-tls=[]
-grpc=[]
+default = ["grpc", "eventmesh_message"]
+full = ["grpc", "eventmesh_message","cloud_events"]
+eventmesh_message = []
+cloud_events = []
+tls = []
+grpc = []
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [dependencies]
@@ -57,6 +58,9 @@ serde_json = "1.0"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 
+#cloudEvents
+cloudevents-sdk = "0.7.0"
+
 # tools crate
 thiserror = "1.0"
 bytes = "1"
@@ -65,16 +69,17 @@ uuid = { version = "1.4.1", features = ["v4"] }
 local-ip-address = "0.5.6"
 futures = "0.3"
 log = "0.4.20"
+chrono = "0.4"
 
 [build-dependencies]
 tonic-build = "0.10"
 
 [[example]]
 name = "producer_example"
 path = "examples/grpc/producer_example.rs"
-required-features = ["grpc","eventmesh_message"]
+required-features = ["grpc", "eventmesh_message","cloud_events"]
 
 [[example]]
 name = "consumer_example"
 path = "examples/grpc/consumer_example.rs"
-required-features = ["grpc","eventmesh_message"]
+required-features = ["grpc", "eventmesh_message"]
diff --git a/eventmesh-sdks/eventmesh-sdk-rust/examples/grpc/producer_example.rs b/eventmesh-sdks/eventmesh-sdk-rust/examples/grpc/producer_example.rs
@@ -15,47 +15,88 @@
  * limitations under the License.
  */
 use std::time::{SystemTime, UNIX_EPOCH};
+
+use chrono::Utc;
+use cloudevents::{EventBuilder, EventBuilderV10};
 use tracing::info;
 
+use eventmesh::common::ProtocolKey;
 use eventmesh::config::EventMeshGrpcClientConfig;
 use eventmesh::grpc::grpc_producer::EventMeshGrpcProducer;
-use eventmesh::grpc::GrpcEventMeshMessageProducer;
+use eventmesh::grpc::GrpcEventMeshProducer;
 use eventmesh::log;
 use eventmesh::model::message::EventMeshMessage;
 
 #[eventmesh::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     log::init_logger();
 
-    let grpc_client_config = EventMeshGrpcClientConfig::new();
-    let mut producer = GrpcEventMeshMessageProducer::new(grpc_client_config);
-
     //Publish Message
-    info!("Publish Message to EventMesh........");
-    let message = EventMeshMessage::default()
-        .with_biz_seq_no("1")
-        .with_content("123")
-        .with_create_time(SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64)
-        .with_topic("123")
-        .with_unique_id("1111");
-    let response = producer.publish(message.clone()).await?;
-    info!("Publish Message to EventMesh return result: {}", response);
-
-    //Publish batch message
-    info!("Publish batch message to EventMesh........");
-    let messages = vec![message.clone(), message.clone(), message.clone()];
-    let response = producer.publish_batch(messages).await?;
-    info!(
-        "Publish batch message to EventMesh return result: {}",
-        response
-    );
-
-    //Publish batch message
-    info!("Publish request reply message to EventMesh........");
-    let response = producer.request_reply(message.clone(), 1000).await?;
-    info!(
-        "Publish request reply message to EventMesh return result: {}",
-        response
-    );
+    #[cfg(feature = "eventmesh_message")]
+    {
+        let grpc_client_config = EventMeshGrpcClientConfig::new();
+        let mut producer = GrpcEventMeshProducer::new(grpc_client_config);
+        info!("Publish Message to EventMesh........");
+        let message = EventMeshMessage::default()
+            .with_biz_seq_no("1")
+            .with_content("123")
+            .with_create_time(SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis() as u64)
+            .with_topic("123")
+            .with_unique_id("1111");
+        let response = producer.publish(message.clone()).await?;
+        info!("Publish Message to EventMesh return result: {}", response);
+
+        //Publish batch message
+        info!("Publish batch message to EventMesh........");
+        let messages = vec![message.clone(), message.clone(), message.clone()];
+        let response = producer.publish_batch(messages).await?;
+        info!(
+            "Publish batch message to EventMesh return result: {}",
+            response
+        );
+
+        //Publish batch message
+        info!("Publish request reply message to EventMesh........");
+        let response = producer.request_reply(message.clone(), 1000).await?;
+        info!(
+            "Publish request reply message to EventMesh return result: {}",
+            response
+        );
+    }
+
+    #[cfg(feature = "cloud_events")]
+    {
+        let grpc_client_config = EventMeshGrpcClientConfig::new();
+        let mut producer = GrpcEventMeshProducer::new(grpc_client_config);
+        info!("Publish Message to EventMesh........");
+        let message = EventBuilderV10::new()
+            .id("my_event.my_application")
+            .source("http://localhost:8080")
+            .subject("mxsm")
+            .ty("example.demo")
+            .time(Utc::now())
+            .data(ProtocolKey::CLOUDEVENT_CONTENT_TYPE, "{\"aaa\":\"1111\"}")
+            .build()?;
+        let response = producer.publish(message.clone()).await?;
+        info!("Publish Message to EventMesh return result: {}", response);
+
+        //Publish batch message
+        info!("Publish batch message to EventMesh........");
+        let messages = vec![message.clone(), message.clone(), message.clone()];
+        let response = producer.publish_batch(messages).await?;
+        info!(
+            "Publish batch message to EventMesh return result: {}",
+            response
+        );
+
+        //Publish batch message
+        info!("Publish request reply message to EventMesh........");
+        let response = producer.request_reply(message.clone(), 1000).await?;
+        info!(
+            "Publish request reply message to EventMesh return result: {}",
+            response
+        );
+    }
+
     Ok(())
 }
diff --git a/eventmesh-sdks/eventmesh-sdk-rust/rust-toolchain.toml b/eventmesh-sdks/eventmesh-sdk-rust/rust-toolchain.toml
@@ -18,4 +18,4 @@
 
 [toolchain]
 # TODO: we can remove this toolchain file when AFIT and RPITIT hits stable.
-channel = "nightly"
+channel = "nightly-x86_64-pc-windows-msvc"
diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/common.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/common.rs
@@ -20,7 +20,7 @@
 pub mod constants;
 
 /// Eventmesh message utilities.
-pub mod eventmesh_message_utils;
+pub mod grpc_eventmesh_message_utils;
 
 /// Local IP helper.
 pub(crate) mod local_ip;