cloudevents/eventmesh message protocol pub/sub for sdk in rocketmq-co…

…nnector (apache#629) * fix cloudevents/eventmesh message protocol pub/sub for sdk in rocketmq-connector
xwm1992 · Dec 7, 2021 · 3b82ce6 · 3b82ce6
1 parent eaef797
commit 3b82ce6
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/.../main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/RocketMQMessageFactory.java b/.../main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/RocketMQMessageFactory.java
@@ -53,8 +53,8 @@ public static MessageReader createReader(final Map<String, String> props,
         throws CloudEventRWException {
 
         return MessageUtils.parseStructuredOrBinaryMessage(
-            () -> props.get(RocketMQHeaders.CONTENT_TYPE),
-            format -> new GenericStructuredMessageReader(format, body),
+            () -> null,
+            format -> null,
             () -> props.get(RocketMQHeaders.SPEC_VERSION),
             sv -> new RocketMQBinaryMessageReader(sv, props, body)
         );

diff --git a/...es/src/main/java/org/apache/eventmesh/http/demo/pub/cloudevents/AsyncPublishInstance.java b/...es/src/main/java/org/apache/eventmesh/http/demo/pub/cloudevents/AsyncPublishInstance.java
@@ -70,7 +70,10 @@ public static void main(String[] args) throws Exception {
             .idc("idc")
             .ip(IPUtils.getLocalAddress())
             .sys("1234")
-            .pid(String.valueOf(ThreadUtils.getPID())).build();
+            .pid(String.valueOf(ThreadUtils.getPID()))
+            .userName("eventmesh")
+            .password("pass")
+            .build();
 
         try (EventMeshHttpProducer eventMeshHttpProducer = new EventMeshHttpProducer(eventMeshClientConfig)) {
             for (int i = 0; i < messageSize; i++) {