[ISSUE apache#4625]fix grpc request reply can not work

mxsm · Dec 7, 2023 · e84412f · e84412f
1 parent 8c59c7f
commit e84412f
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 4 deletions.
diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/common/protocol_key.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/common/protocol_key.rs
@@ -44,6 +44,8 @@ impl ProtocolKey {
     pub const GRPC_RESPONSE_MESSAGE: &'static str = "response_message";
     pub const GRPC_RESPONSE_TIME: &'static str = "time";
 
+    pub const SUB_MESSAGE_TYPE: &'static str = "submessagetype";
+
     // CloudEvents spec
     pub const ID: &'static str = "id";
     pub const SOURCE: &'static str = "source";

diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/grpc/grpc_consumer.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/grpc/grpc_consumer.rs
@@ -157,6 +157,14 @@ impl EventMeshGrpcConsumer {
                             None,
                             properties,
                         );
+                        received.attributes.insert(
+                            ProtocolKey::SUB_MESSAGE_TYPE.to_string(),
+                            PbCloudEventAttributeValue {
+                                attr: Some(PbAttr::CeString(
+                                    SubscriptionReply::SUB_TYPE.to_string(),
+                                )),
+                            },
+                        );
                         received.data =
                             Some(PbData::TextData(serde_json::to_string(&reply).unwrap()));
                         let _ = keeper.sender.send(received).await;

diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/model/message.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/model/message.rs
@@ -17,13 +17,13 @@
 
 #![cfg(feature = "eventmesh_message")]
 
+#[allow(unused_imports)]
+use cloudevents::Event;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt;
 use std::time::{SystemTime, UNIX_EPOCH};
 
-use cloudevents::Event;
-use serde::{Deserialize, Serialize};
-
 use crate::common::grpc_eventmesh_message_utils::EventMeshCloudEventUtils;
 use crate::model::convert::FromPbCloudEvent;
 use crate::proto_cloud_event::PbCloudEvent;

diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/model/subscription.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/model/subscription.rs
@@ -201,6 +201,8 @@ pub struct SubscriptionReply {
 }
 
 impl SubscriptionReply {
+    pub const SUB_TYPE: &'static str = "subscription_reply";
+
     pub fn new(
         producer_group: String,
         topic: String,

diff --git a/eventmesh-sdks/eventmesh-sdk-rust/src/net.rs b/eventmesh-sdks/eventmesh-sdk-rust/src/net.rs
@@ -14,7 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
+#![allow(unused_imports)]
 pub use crate::net::grpc::grpc_client::GrpcClient;
 pub use crate::net::grpc::grpc_client::SubscribeStreamKeeper;
 mod grpc;