connector support cloud event (#586)

Co-authored-by: wangshaojie <[email protected]>

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
+ *
+ * https://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
+ *
+ * https://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
+ *
+ * https://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
+ *
+ * https://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;
+ }
+}