support cloudEvents

apache · xwm1992 · Nov 19, 2021 · Oct 28, 2021 · Nov 1, 2021 · Nov 1, 2021
commit f0f31086329cfcbe517e8210020e1e5c5ea2da85
diff --git a/...sh-connector-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java b/...sh-connector-api/src/main/java/org/apache/eventmesh/api/EventMeshAsyncConsumeContext.java
@@ -17,8 +17,6 @@
 
 package org.apache.eventmesh.api;
 
-import io.openmessaging.api.Action;
-import io.openmessaging.api.AsyncConsumeContext;
 
 public abstract class EventMeshAsyncConsumeContext extends AsyncConsumeContext {
 
@@ -34,8 +32,4 @@ public void setAbstractContext(AbstractContext abstractContext) {
 
     public abstract void commit(EventMeshAction action);
 
-    @Override
-    public void commit(Action action) {
-        throw new UnsupportedOperationException("not support yet");
-    }
 }
diff --git a/.../eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/RequestReplyCallback.java b/.../eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/RequestReplyCallback.java
@@ -0,0 +1,10 @@
+package org.apache.eventmesh.api;
+
+import io.cloudevents.CloudEvent;
+
+public interface RequestReplyCallback {
+
+    void onSuccess(CloudEvent event);
+
+    void onException(Throwable e);
+}
diff --git a/...gin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/producer/Producer.java b/...gin/eventmesh-connector-api/src/main/java/org/apache/eventmesh/api/producer/Producer.java
@@ -17,10 +17,7 @@
 
 package org.apache.eventmesh.api.producer;
 
-import org.apache.eventmesh.api.LifeCycle;
-import org.apache.eventmesh.api.RRCallback;
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.*;
 import org.apache.eventmesh.spi.EventMeshExtensionType;
 import org.apache.eventmesh.spi.EventMeshSPI;
 
@@ -44,7 +41,7 @@ public interface Producer extends LifeCycle {
 
     void sendAsync(final CloudEvent cloudEvent, final SendCallback sendCallback);
 
-    void request(CloudEvent cloudEvent, RRCallback rrCallback, long timeout) throws Exception;
+    void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout) throws Exception;
 
     boolean reply(final CloudEvent cloudEvent, final SendCallback sendCallback) throws Exception;
 

diff --git a/...h-connector-rocketmq/src/test/java/org/apache/rocketmq/consumer/PushConsumerImplTest.java b/...h-connector-rocketmq/src/test/java/org/apache/rocketmq/consumer/PushConsumerImplTest.java
@@ -23,13 +23,7 @@
 import java.util.Collections;
 import java.util.Properties;
 
-import io.openmessaging.api.AsyncConsumeContext;
-import io.openmessaging.api.AsyncMessageListener;
-import io.openmessaging.api.Consumer;
-import io.openmessaging.api.Message;
-import io.openmessaging.api.MessagingAccessPoint;
-import io.openmessaging.api.OMS;
-import io.openmessaging.api.OMSBuiltinKeys;
+import io.openmessaging.api.*;
 
 import org.apache.eventmesh.api.EventMeshAction;
 import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;

diff --git a/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle b/eventmesh-protocol-plugin/eventmesh-protocol-api/build.gradle
@@ -17,7 +17,7 @@
 
 dependencies {
     implementation project(":eventmesh-spi")
-    implementation project(":eventmesh-common")
+    api project(":eventmesh-common")
 
     implementation "io.cloudevents:cloudevents-core"
 

diff --git a/...entmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java b/...entmesh-protocol-api/src/main/java/org/apache/eventmesh/protocol/api/ProtocolAdaptor.java
@@ -17,6 +17,7 @@
 
 package org.apache.eventmesh.protocol.api;
 
+import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
 import org.apache.eventmesh.spi.EventMeshExtensionType;
 import org.apache.eventmesh.spi.EventMeshSPI;
@@ -32,15 +33,15 @@
  * @since 1.3.0
  */
 @EventMeshSPI(isSingleton = true, eventMeshExtensionType = EventMeshExtensionType.PROTOCOL)
-public interface ProtocolAdaptor {
+public interface ProtocolAdaptor<T> {
 
     /**
      * transform protocol to {@link CloudEvent}.
      *
      * @param protocol input protocol
      * @return cloud event
      */
-    CloudEventV1 toCloudEventV1(Package protocol) throws ProtocolHandleException;
+    CloudEventV1 toCloudEventV1(T protocol) throws ProtocolHandleException;
 
 
     /**

diff --git a/...s/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java b/...s/src/main/java/org/apache/eventmesh/protocol/cloudevents/CloudEventsProtocolAdaptor.java
@@ -17,6 +17,8 @@
 
 package org.apache.eventmesh.protocol.cloudevents;
 
+import org.apache.eventmesh.common.command.HttpCommand;
+import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
 
 import io.cloudevents.core.v1.CloudEventV1;
@@ -26,10 +28,16 @@
  *
  * @since 1.3.0
  */
-public class CloudEventsProtocolAdaptor implements ProtocolAdaptor {
+public class CloudEventsProtocolAdaptor<T> implements ProtocolAdaptor<T> {
 
     @Override
-    public CloudEventV1 toCloudEventV1(Package cloudEvent) {
+    public CloudEventV1 toCloudEventV1(T cloudEvent) {
+
+        if (cloudEvent instanceof Package){
+            //todo:convert package to cloudevents
+        }else if (cloudEvent instanceof HttpCommand){
+            //todo:convert httpCommand to cloudevents
+        }
         return null;
     }
 

diff --git a/...e/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java b/...e/src/main/java/org/apache/eventmesh/protocol/openmessage/OpenMessageProtocolAdaptor.java
@@ -17,6 +17,7 @@
 
 package org.apache.eventmesh.protocol.openmessage;
 
+import org.apache.eventmesh.common.protocol.tcp.Package;
 import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
 
 import io.cloudevents.CloudEvent;
@@ -29,10 +30,10 @@
  *
  * @since 1.3.0
  */
-public class OpenMessageProtocolAdaptor implements ProtocolAdaptor {
+public class OpenMessageProtocolAdaptor<T> implements ProtocolAdaptor<T> {
 
     @Override
-    public CloudEventV1 toCloudEventV1(Package message) {
+    public CloudEventV1 toCloudEventV1(T message) {
         return null;
     }
 

diff --git a/...esh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQProducerWrapper.java b/...esh-runtime/src/main/java/org/apache/eventmesh/runtime/core/plugin/MQProducerWrapper.java
@@ -20,6 +20,7 @@
 import java.util.Properties;
 
 import org.apache.eventmesh.api.RRCallback;
+import org.apache.eventmesh.api.RequestReplyCallback;
 import org.apache.eventmesh.api.SendCallback;
 import org.apache.eventmesh.api.factory.ConnectorPluginFactory;
 import org.apache.eventmesh.api.producer.Producer;
@@ -80,7 +81,7 @@ public void send(CloudEvent cloudEvent, SendCallback sendCallback) throws Except
         meshMQProducer.publish(cloudEvent, sendCallback);
     }
 
-    public void request(CloudEvent cloudEvent, RRCallback rrCallback, long timeout)
+    public void request(CloudEvent cloudEvent, RequestReplyCallback rrCallback, long timeout)
             throws Exception {
         meshMQProducer.request(cloudEvent, rrCallback, timeout);
     }

diff --git a/...main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java b/...main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/EventMeshConsumer.java
@@ -23,6 +23,7 @@
 
 import io.cloudevents.CloudEvent;
 
+import io.cloudevents.core.v1.CloudEventBuilder;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.eventmesh.api.*;
 import org.apache.eventmesh.api.exception.OnExceptionContext;
@@ -109,14 +110,17 @@ public void subscribe(String topic, SubscriptionItem subscriptionItem) throws Ex
             listener = new EventListener() {
                 @Override
                 public void consume(CloudEvent event, AsyncConsumeContext context) {
-                    String topic = event.getExtension(Constants.PROPERTY_MESSAGE_DESTINATION);
+                    String topic = event.getSubject();
 //                    String topic = message.getSystemProperties(Constants.PROPERTY_MESSAGE_DESTINATION);
-                    String bizSeqNo = message.getSystemProperties(Constants.PROPERTY_MESSAGE_SEARCH_KEYS);
-                    String uniqueId = message.getUserProperties(Constants.RMB_UNIQ_ID);
+                    String bizSeqNo = (String) event.getExtension(Constants.PROPERTY_MESSAGE_SEARCH_KEYS);
+                    String uniqueId = (String) event.getExtension(Constants.RMB_UNIQ_ID);
 
-                    message.getUserProperties().put(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()));
+                    event = new CloudEventBuilder(event)
+                            .withExtension(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
+                            .build();
+//                    message.getUserProperties().put(EventMeshConstants.REQ_MQ2EVENTMESH_TIMESTAMP, String.valueOf(System.currentTimeMillis()));
                     if (messageLogger.isDebugEnabled()) {
-                        messageLogger.debug("message|mq2eventMesh|topic={}|msg={}", topic, message);
+                        messageLogger.debug("message|mq2eventMesh|topic={}|event={}", topic, event);
                     } else {
                         messageLogger.info("message|mq2eventMesh|topic={}|bizSeqNo={}|uniqueId={}", topic, bizSeqNo, uniqueId);
                     }
@@ -127,7 +131,7 @@ public void consume(CloudEvent event, AsyncConsumeContext context) {
                     if (currentTopicConfig == null) {
                         logger.error("no topicConfig found, consumerGroup:{} topic:{}", consumerGroupConf.getConsumerGroup(), topic);
                         try {
-                            sendMessageBack(message, uniqueId, bizSeqNo);
+                            sendMessageBack(event, uniqueId, bizSeqNo);
 //                            context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
 //                            context.ack();
                             eventMeshAsyncConsumeContext.commit(EventMeshAction.CommitMessage);
@@ -136,15 +140,15 @@ public void consume(CloudEvent event, AsyncConsumeContext context) {
                         }
                     }
                     HandleMsgContext handleMsgContext = new HandleMsgContext(EventMeshUtil.buildPushMsgSeqNo(), consumerGroupConf.getConsumerGroup(), EventMeshConsumer.this,
-                            topic, message, subscriptionItem, eventMeshAsyncConsumeContext.getAbstractContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
+                            topic, event, subscriptionItem, eventMeshAsyncConsumeContext.getAbstractContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
 
                     if (httpMessageHandler.handle(handleMsgContext)) {
 //                        context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_FINISH.name());
 //                        context.ack();
                         eventMeshAsyncConsumeContext.commit(EventMeshAction.ManualAck);
                     } else {
                         try {
-                            sendMessageBack(message, uniqueId, bizSeqNo);
+                            sendMessageBack(event, uniqueId, bizSeqNo);
                         } catch (Exception e) {
 
                         }
@@ -177,7 +181,7 @@ public void consume(CloudEvent event, AsyncConsumeContext context) {
                     if (currentTopicConfig == null) {
                         logger.error("no topicConfig found, consumerGroup:{} topic:{}", consumerGroupConf.getConsumerGroup(), topic);
                         try {
-                            sendMessageBack(message, uniqueId, bizSeqNo);
+                            sendMessageBack(event, uniqueId, bizSeqNo);
 //                            context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_SUCCESS.name());
 //                            context.ack();
                             eventMeshAsyncConsumeContext.commit(EventMeshAction.CommitMessage);
@@ -186,15 +190,15 @@ public void consume(CloudEvent event, AsyncConsumeContext context) {
                         }
                     }
                     HandleMsgContext handleMsgContext = new HandleMsgContext(EventMeshUtil.buildPushMsgSeqNo(), consumerGroupConf.getConsumerGroup(), EventMeshConsumer.this,
-                            topic, message, subscriptionItem, eventMeshAsyncConsumeContext.getAbstractContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
+                            topic, event, subscriptionItem, eventMeshAsyncConsumeContext.getAbstractContext(), consumerGroupConf, eventMeshHTTPServer, bizSeqNo, uniqueId, currentTopicConfig);
 
                     if (httpMessageHandler.handle(handleMsgContext)) {
 //                        context.attributes().put(NonStandardKeys.MESSAGE_CONSUME_STATUS, EventMeshConsumeConcurrentlyStatus.CONSUME_FINISH.name());
 //                        context.ack();
                         eventMeshAsyncConsumeContext.commit(EventMeshAction.ManualAck);
                     } else {
                         try {
-                            sendMessageBack(message, uniqueId, bizSeqNo);
+                            sendMessageBack(event, uniqueId, bizSeqNo);
                         } catch (Exception e) {
 
                         }
@@ -248,7 +252,7 @@ public EventMeshHTTPServer getEventMeshHTTPServer() {
         return eventMeshHTTPServer;
     }
 
-    public void sendMessageBack(final Message msgBack, final String uniqueId, String bizSeqNo) throws Exception {
+    public void sendMessageBack(final CloudEvent event, final String uniqueId, String bizSeqNo) throws Exception {
 
         EventMeshProducer sendMessageBack
                 = eventMeshHTTPServer.getProducerManager().getEventMeshProducer(consumerGroupConf.getConsumerGroup());
@@ -258,7 +262,7 @@ public void sendMessageBack(final Message msgBack, final String uniqueId, String
             return;
         }
 
-        final SendMessageContext sendMessageBackContext = new SendMessageContext(bizSeqNo, msgBack, sendMessageBack, eventMeshHTTPServer);
+        final SendMessageContext sendMessageBackContext = new SendMessageContext(bizSeqNo, event, sendMessageBack, eventMeshHTTPServer);
 
         sendMessageBack.send(sendMessageBackContext, new SendCallback() {
             @Override

diff --git a/.../main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HandleMsgContext.java b/.../main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/HandleMsgContext.java
@@ -21,8 +21,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import io.openmessaging.api.Message;
-
+import io.cloudevents.CloudEvent;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.time.DateFormatUtils;
 import org.apache.eventmesh.api.AbstractContext;
@@ -53,7 +52,7 @@ public class HandleMsgContext {
 
     private SubscriptionItem subscriptionItem;
 
-    private Message msg;
+    private CloudEvent event;
 
     private int ttl;
 
@@ -70,22 +69,23 @@ public class HandleMsgContext {
     private Map<String, String> props;
 
     public HandleMsgContext(String msgRandomNo, String consumerGroup, EventMeshConsumer eventMeshConsumer,
-                            String topic, Message msg, SubscriptionItem subscriptionItem,
+                            String topic, CloudEvent event, SubscriptionItem subscriptionItem,
                             AbstractContext context, ConsumerGroupConf consumerGroupConfig,
                             EventMeshHTTPServer eventMeshHTTPServer, String bizSeqNo, String uniqueId, ConsumerGroupTopicConf consumeTopicConfig) {
         this.msgRandomNo = msgRandomNo;
         this.consumerGroup = consumerGroup;
         this.eventMeshConsumer = eventMeshConsumer;
         this.topic = topic;
-        this.msg = msg;
+        this.event = event;
         this.subscriptionItem = subscriptionItem;
         this.context = context;
         this.consumerGroupConfig = consumerGroupConfig;
         this.eventMeshHTTPServer = eventMeshHTTPServer;
         this.bizSeqNo = bizSeqNo;
         this.uniqueId = uniqueId;
         this.consumeTopicConfig = consumeTopicConfig;
-        String ttlStr = msg.getUserProperties(Constants.PROPERTY_MESSAGE_TIMEOUT);
+
+        String ttlStr = (String) event.getExtension(Constants.PROPERTY_MESSAGE_TIMEOUT);
         this.ttl = StringUtils.isNumeric(ttlStr)? Integer.parseInt(ttlStr): EventMeshConstants.DEFAULT_TIMEOUT_IN_MILLISECONDS;
     }
 
@@ -148,12 +148,12 @@ public void setTopic(String topic) {
         this.topic = topic;
     }
 
-    public Message getMsg() {
-        return msg;
+    public CloudEvent getEvent() {
+        return event;
     }
 
-    public void setMsg(Message msg) {
-        this.msg = msg;
+    public void setEvent(CloudEvent event) {
+        this.event = event;
     }
 
     public SubscriptionItem getSubscriptionItem() {

diff --git a/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java b/...java/org/apache/eventmesh/runtime/core/protocol/http/processor/ReplyMessageProcessor.java
@@ -207,7 +207,9 @@ public void onResponse(HttpCommand httpCommand) {
 //                replyMessageRequestBody.getContent());
 
         try {
-            CloudEvent clone = new CloudEventBuilder(sendMessageContext.getEvent()).withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis());
+            CloudEvent clone = new CloudEventBuilder(sendMessageContext.getEvent())
+                    .withExtension(EventMeshConstants.REQ_EVENTMESH2MQ_TIMESTAMP, String.valueOf(System.currentTimeMillis()))
+                    .build();
             sendMessageContext.setEvent(clone);
             eventMeshProducer.reply(sendMessageContext, new SendCallback() {
                 @Override