connector support cloud event

eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/MessagingAccessPointImpl.java

@@ -29,9 +29,6 @@
 import io.openmessaging.api.transaction.LocalTransactionChecker;
 import io.openmessaging.api.transaction.TransactionProducer;
 
-import org.apache.eventmesh.connector.rocketmq.consumer.PushConsumerImpl;
-import org.apache.eventmesh.connector.rocketmq.producer.ProducerImpl;
-
 public class MessagingAccessPointImpl implements MessagingAccessPoint {
 
     private Properties accessPointProperties;
@@ -52,7 +49,7 @@ public Properties attributes() {
 
     @Override
     public Producer createProducer(Properties properties) {
-        return new ProducerImpl(this.accessPointProperties);
+        return null;
     }
 
     @Override
@@ -72,7 +69,7 @@ public TransactionProducer createTransactionProducer(Properties properties) {
 
     @Override
     public Consumer createConsumer(Properties properties) {
-        return new PushConsumerImpl(properties);
+        return null;
     }
 
     @Override

eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/RocketMQMessageFactory.java

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rocketmq.cloudevent;
+
+import org.apache.rocketmq.common.message.Message;
+
+import java.util.Map;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import io.cloudevents.core.message.MessageReader;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.core.message.impl.GenericStructuredMessageReader;
+import io.cloudevents.core.message.impl.MessageUtils;
+import io.cloudevents.lang.Nullable;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+
+import org.apache.eventmesh.connector.rocketmq.cloudevent.impl.RocketMQBinaryMessageReader;
+import org.apache.eventmesh.connector.rocketmq.cloudevent.impl.RocketMQHeaders;
+import org.apache.eventmesh.connector.rocketmq.cloudevent.impl.RocketMQMessageWriter;
+
+
+@ParametersAreNonnullByDefault
+public final class RocketMQMessageFactory {
+
+    private RocketMQMessageFactory() {
+        // prevent instantiation
+    }
+
+    public static MessageReader createReader(final Message message) throws CloudEventRWException {
+        return createReader(message.getProperties(), message.getBody());
+    }
+
+
+    public static MessageReader createReader(final Map<String, String> props,
+                                             @Nullable final byte[] body)
+        throws CloudEventRWException {
+
+        return MessageUtils.parseStructuredOrBinaryMessage(
+            () -> props.get(RocketMQHeaders.CONTENT_TYPE),
+            format -> new GenericStructuredMessageReader(format, body),
+            () -> props.get(RocketMQHeaders.SPEC_VERSION),
+            sv -> new RocketMQBinaryMessageReader(sv, props, body)
+        );
+    }
+
+
+    public static MessageWriter<CloudEventWriter<Message>, Message> createWriter(String topic) {
+        return new RocketMQMessageWriter<>(topic);
+    }
+
+    public static MessageWriter<CloudEventWriter<Message>, Message> createWriter(String topic,
+                                                                                 String keys) {
+        return new RocketMQMessageWriter<>(topic, keys);
+    }
+
+    public static MessageWriter<CloudEventWriter<Message>, Message> createWriter(String topic,
+                                                                                 String keys,
+                                                                                 String tags) {
+        return new RocketMQMessageWriter<>(topic, keys, tags);
+    }
+
+}

eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/impl/RocketMQBinaryMessageReader.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rocketmq.cloudevent.impl;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.data.BytesCloudEventData;
+import io.cloudevents.core.message.impl.BaseGenericBinaryMessageReaderImpl;
+
+public class RocketMQBinaryMessageReader
+    extends BaseGenericBinaryMessageReaderImpl<String, String> {
+
+    private final Map<String, String> headers;
+
+    public RocketMQBinaryMessageReader(SpecVersion version, Map<String, String> headers,
+                                       byte[] payload) {
+        super(version,
+            payload != null && payload.length > 0 ? BytesCloudEventData.wrap(payload) : null);
+
+        Objects.requireNonNull(headers);
+        this.headers = headers;
+    }
+
+    @Override
+    protected boolean isContentTypeHeader(String key) {
+        return key.equals(RocketMQHeaders.CONTENT_TYPE);
+    }
+
+    @Override
+    protected boolean isCloudEventsHeader(String key) {
+        return key.length() > 3 && key.substring(0, RocketMQHeaders.CE_PREFIX.length())
+            .startsWith(RocketMQHeaders.CE_PREFIX);
+    }
+
+    @Override
+    protected String toCloudEventsKey(String key) {
+        return key.substring(RocketMQHeaders.CE_PREFIX.length()).toLowerCase();
+    }
+
+    @Override
+    protected void forEachHeader(BiConsumer<String, String> fn) {
+        this.headers.forEach((k, v) -> fn.accept(k, v));
+    }
+
+    @Override
+    protected String toCloudEventsValue(String value) {
+        return value;
+    }
+}

eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/impl/RocketMQHeaders.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rocketmq.cloudevent.impl;
+
+import java.util.Map;
+
+import io.cloudevents.core.message.impl.MessageUtils;
+import io.cloudevents.core.v1.CloudEventV1;
+
+public class RocketMQHeaders {
+
+    public static final String CE_PREFIX = "CE_";
+
+    protected static final Map<String, String> ATTRIBUTES_TO_HEADERS =
+        MessageUtils.generateAttributesToHeadersMapping(v -> CE_PREFIX + v);
+
+    public static final String CONTENT_TYPE =
+        ATTRIBUTES_TO_HEADERS.get(CloudEventV1.DATACONTENTTYPE);
+
+    public static final String SPEC_VERSION = ATTRIBUTES_TO_HEADERS.get(CloudEventV1.SPECVERSION);
+
+
+}
+

eventmesh-connector-plugin/eventmesh-connector-rocketmq/src/main/java/org/apache/eventmesh/connector/rocketmq/cloudevent/impl/RocketMQMessageWriter.java

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rocketmq.cloudevent.impl;
+
+import org.apache.rocketmq.common.message.Message;
+
+import io.cloudevents.CloudEventData;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.format.EventFormat;
+import io.cloudevents.core.message.MessageWriter;
+import io.cloudevents.rw.CloudEventContextWriter;
+import io.cloudevents.rw.CloudEventRWException;
+import io.cloudevents.rw.CloudEventWriter;
+
+
+public final class RocketMQMessageWriter<R>
+    implements MessageWriter<CloudEventWriter<Message>, Message>, CloudEventWriter<Message> {
+
+    private Message message;
+
+
+    public RocketMQMessageWriter(String topic) {
+        message = new Message();
+        message.setTopic(topic);
+    }
+
+    public RocketMQMessageWriter(String topic, String keys) {
+        message = new Message();
+
+        message.setTopic(topic);
+
+        if (keys != null && keys.length() > 0) {
+            message.setKeys(keys);
+        }
+    }
+
+    public RocketMQMessageWriter(String topic, String keys, String tags) {
+        message = new Message();
+
+        message.setTopic(topic);
+
+        if (tags != null && tags.length() > 0) {
+            message.setTags(tags);
+        }
+
+        if (keys != null && keys.length() > 0) {
+            message.setKeys(keys);
+        }
+    }
+
+
+    @Override
+    public CloudEventContextWriter withContextAttribute(String name, String value)
+        throws CloudEventRWException {
+
+        String propName = RocketMQHeaders.ATTRIBUTES_TO_HEADERS.get(name);
+        if (propName == null) {
+            propName = RocketMQHeaders.CE_PREFIX + name;
+        }
+        message.putUserProperty(propName, value);
+        return this;
+    }
+
+    @Override
+    public RocketMQMessageWriter<R> create(final SpecVersion version) {
+        message.putUserProperty(RocketMQHeaders.SPEC_VERSION, version.toString());
+        return this;
+    }
+
+    @Override
+    public Message setEvent(final EventFormat format, final byte[] value)
+        throws CloudEventRWException {
+        message.putUserProperty(RocketMQHeaders.CONTENT_TYPE, format.serializedContentType());
+        message.setBody(value);
+        return message;
+    }
+
+    @Override
+    public Message end(final CloudEventData data) throws CloudEventRWException {
+        message.setBody(data.toBytes());
+        return message;
+    }
+
+    @Override
+    public Message end() {
+        message.setBody(null);
+        return message;
+    }
+}