This is an automated email from the ASF dual-hosted git repository.
wqliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new e9010db30 [ISSUE #3377] Add eventmesh-storage-rabbitmq module
new 50f1e6d19 Merge pull request #3378 from xwm1992/add-storage-rabbitmq
e9010db30 is described below
commit e9010db301759b55c4c86db194412800d2aba1ef
Author: xwm1992 <[email protected]>
AuthorDate: Thu Mar 9 16:06:08 2023 +0800
[ISSUE #3377] Add eventmesh-storage-rabbitmq module
---
.../eventmesh-storage-rabbitmq/build.gradle | 48 ++++++
.../eventmesh-storage-rabbitmq/gradle.properties | 18 +++
.../storage/rabbitmq/client/RabbitmqClient.java | 148 ++++++++++++++++++
.../rabbitmq/client/RabbitmqConnectionFactory.java | 40 +++++
.../rabbitmq/cloudevent/RabbitmqCloudEvent.java | 81 ++++++++++
.../cloudevent/RabbitmqCloudEventWriter.java | 68 +++++++++
.../rabbitmq/config/ConfigurationHolder.java | 60 ++++++++
.../rabbitmq/consumer/RabbitmqConsumer.java | 143 ++++++++++++++++++
.../rabbitmq/consumer/RabbitmqConsumerHandler.java | 83 ++++++++++
.../exception/RabbitmqStorageException.java | 31 ++++
.../rabbitmq/producer/RabbitmqProducer.java | 167 +++++++++++++++++++++
.../storage/RabbitmqStorageResourceService.java | 33 ++++
.../storage/rabbitmq/utils/ByteArrayUtils.java | 42 ++++++
.../org.apache.eventmesh.api.consumer.Consumer | 16 ++
.../org.apache.eventmesh.api.producer.Producer | 16 ++
...he.eventmesh.api.storage.StorageResourceService | 16 ++
.../src/main/resources/rabbitmq-client.properties | 31 ++++
.../rabbitmq/RabbitmqMockConnectionFactory.java | 59 ++++++++
.../eventmesh/storage/rabbitmq/RabbitmqServer.java | 55 +++++++
.../cloudevent/RabbitmqCloudEventTest.java | 70 +++++++++
.../rabbitmq/config/ConfigurationHolderTest.java | 62 ++++++++
.../rabbitmq/consumer/RabbitmqConsumerTest.java | 93 ++++++++++++
.../rabbitmq/producer/RabbitmqProducerTest.java | 93 ++++++++++++
.../src/test/resources/rabbitmq-client.properties | 31 ++++
settings.gradle | 1 +
25 files changed, 1505 insertions(+)
diff --git a/eventmesh-storage/eventmesh-storage-rabbitmq/build.gradle
b/eventmesh-storage/eventmesh-storage-rabbitmq/build.gradle
new file mode 100644
index 000000000..65b67c063
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-rabbitmq/build.gradle
@@ -0,0 +1,48 @@
+/*
+ * 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.
+ */
+
+configurations {
+ implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
+ implementation.exclude group: 'log4j', module: 'log4j'
+}
+
+dependencies {
+ implementation project(":eventmesh-storage:eventmesh-storage-api")
+ implementation project(":eventmesh-common")
+ // rabbitmq
+ implementation 'com.rabbitmq:amqp-client:5.16.0'
+
+ testImplementation project(":eventmesh-storage:eventmesh-storage-api")
+ testImplementation project(":eventmesh-common")
+ // rabbitmq
+ testImplementation 'com.rabbitmq:amqp-client:5.16.0'
+
+ implementation 'io.cloudevents:cloudevents-json-jackson'
+ testImplementation 'io.cloudevents:cloudevents-json-jackson'
+
+ testImplementation "org.mockito:mockito-core"
+ testImplementation "org.powermock:powermock-module-junit4"
+ testImplementation "org.powermock:powermock-api-mockito2"
+
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+
+ testImplementation('com.github.fridujo:rabbitmq-mock:1.1.1')
+}
diff --git a/eventmesh-storage/eventmesh-storage-rabbitmq/gradle.properties
b/eventmesh-storage/eventmesh-storage-rabbitmq/gradle.properties
new file mode 100644
index 000000000..bbf99dcbf
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-rabbitmq/gradle.properties
@@ -0,0 +1,18 @@
+# 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.
+#
+
+pluginType=storage
+pluginName=rabbitmq
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqClient.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqClient.java
new file mode 100644
index 000000000..b9b0690ce
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqClient.java
@@ -0,0 +1,148 @@
+/*
+ * 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.storage.rabbitmq.client;
+
+import org.apache.commons.lang3.StringUtils;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RabbitmqClient {
+
+
+ private final RabbitmqConnectionFactory rabbitmqConnectionFactory;
+
+ public RabbitmqClient(RabbitmqConnectionFactory rabbitmqConnectionFactory)
{
+ this.rabbitmqConnectionFactory = rabbitmqConnectionFactory;
+ }
+
+
+ /**
+ * get rabbitmq connection
+ *
+ * @param host host
+ * @param username username
+ * @param passwd password
+ * @param port port
+ * @param virtualHost virtual host
+ * @return connection
+ * @throws Exception Exception
+ */
+ public Connection getConnection(String host, String username,
+ String passwd, int port,
+ String virtualHost) throws Exception {
+ ConnectionFactory factory =
rabbitmqConnectionFactory.createConnectionFactory();
+ factory.setHost(host.trim());
+ factory.setPort(port);
+ if (StringUtils.isNotEmpty(virtualHost)) {
+ factory.setVirtualHost(virtualHost.trim());
+ }
+ factory.setUsername(username);
+ factory.setPassword(passwd.trim());
+
+ return rabbitmqConnectionFactory.createConnection(factory);
+ }
+
+ /**
+ * send message
+ *
+ * @param channel channel
+ * @param exchangeName exchange name
+ * @param routingKey routing key
+ * @param message message
+ * @throws Exception Exception
+ */
+ public void publish(Channel channel, String exchangeName,
+ String routingKey, byte[] message) throws Exception {
+ channel.basicPublish(exchangeName, routingKey, null, message);
+ }
+
+ /**
+ * binding queue
+ *
+ * @param channel channel
+ * @param builtinExchangeType exchange type
+ * @param exchangeName exchange name
+ * @param routingKey routing key
+ * @param queueName queue name
+ */
+ public void binding(Channel channel, BuiltinExchangeType
builtinExchangeType,
+ String exchangeName, String routingKey, String queueName) {
+ try {
+ channel.exchangeDeclare(exchangeName,
builtinExchangeType.getType(), true,
+ false, false, null);
+ channel.queueDeclare(queueName, false, false,
+ false, null);
+ routingKey =
builtinExchangeType.getType().equals(BuiltinExchangeType.FANOUT.getType()) ? ""
: routingKey;
+ channel.queueBind(queueName, exchangeName, routingKey);
+ } catch (Exception ex) {
+ log.error("[RabbitmqClient] binding happen exception.", ex);
+ }
+ }
+
+ /**
+ * unbinding queue
+ *
+ * @param channel channel
+ * @param exchangeName exchange name
+ * @param routingKey routing key
+ * @param queueName queue name
+ */
+ public void unbinding(Channel channel, String exchangeName, String
routingKey, String queueName) {
+ try {
+ channel.queueUnbind(queueName, exchangeName, routingKey);
+ } catch (Exception ex) {
+ log.error("[RabbitmqClient] unbinding happen exception.", ex);
+ }
+ }
+
+ /**
+ * close connection
+ *
+ * @param connection connection
+ */
+ public void closeConnection(Connection connection) {
+ if (null != connection) {
+ try {
+ connection.close();
+ } catch (Exception ex) {
+ log.error("[RabbitmqClient] connection close happen
exception.", ex);
+ }
+ }
+ }
+
+ /**
+ * close channel
+ *
+ * @param channel channel
+ */
+ public void closeChannel(Channel channel) {
+ if (null != channel) {
+ try {
+ channel.close();
+ } catch (Exception ex) {
+ log.error("[RabbitmqClient] channel close happen exception.",
ex);
+ }
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqConnectionFactory.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqConnectionFactory.java
new file mode 100644
index 000000000..f3866a628
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/client/RabbitmqConnectionFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.storage.rabbitmq.client;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+public class RabbitmqConnectionFactory {
+
+ public ConnectionFactory createConnectionFactory() {
+ return new ConnectionFactory();
+ }
+
+ public Connection createConnection(ConnectionFactory connectionFactory)
throws IOException, TimeoutException {
+ return connectionFactory.newConnection();
+ }
+
+ public Channel createChannel(Connection connection) throws IOException {
+ return connection.createChannel();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEvent.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEvent.java
new file mode 100644
index 000000000..8491d8947
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEvent.java
@@ -0,0 +1,81 @@
+/*
+ * 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.storage.rabbitmq.cloudevent;
+
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.JsonUtils;
+import
org.apache.eventmesh.storage.rabbitmq.exception.RabbitmqStorageException;
+import org.apache.eventmesh.storage.rabbitmq.utils.ByteArrayUtils;
+
+import java.io.Serializable;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.SpecVersion;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class RabbitmqCloudEvent implements Serializable {
+
+ private SpecVersion version;
+ private String data;
+ private Map<String, String> extensions = new HashMap<>();
+
+ public CloudEvent convertToCloudEvent() {
+ CloudEventBuilder builder;
+ switch (version) {
+ case V03:
+ builder = CloudEventBuilder.v03();
+ break;
+ case V1:
+ builder = CloudEventBuilder.v1();
+ break;
+ default:
+ throw new RabbitmqStorageException(String.format("CloudEvent
version %s does not support.", version));
+ }
+ builder.withData(data.getBytes(StandardCharsets.UTF_8))
+ .withId(extensions.remove("id"))
+ .withSource(URI.create(extensions.remove("source")))
+ .withType(extensions.remove("type"))
+ .withDataContentType(extensions.remove("datacontenttype"))
+ .withSubject(extensions.remove("subject"));
+ extensions.forEach(builder::withExtension);
+
+ return builder.build();
+ }
+
+ public static byte[] toByteArray(RabbitmqCloudEvent rabbitmqCloudEvent)
throws Exception {
+ Optional<byte[]> optionalBytes =
ByteArrayUtils.objectToBytes(rabbitmqCloudEvent);
+ return optionalBytes.orElseGet(() -> new byte[]{});
+ }
+
+ public static RabbitmqCloudEvent getFromByteArray(byte[] body) {
+ return JsonUtils.parseTypeReferenceObject(new String(body,
Constants.DEFAULT_CHARSET), new TypeReference<RabbitmqCloudEvent>() {
+ });
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventWriter.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventWriter.java
new file mode 100644
index 000000000..fef86736e
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventWriter.java
@@ -0,0 +1,68 @@
+/*
+ * 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.storage.rabbitmq.cloudevent;
+
+import java.nio.charset.StandardCharsets;
+
+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 class RabbitmqCloudEventWriter
+ implements MessageWriter<CloudEventWriter<RabbitmqCloudEvent>,
RabbitmqCloudEvent>, CloudEventWriter<RabbitmqCloudEvent> {
+
+ private final RabbitmqCloudEvent rabbitmqCloudEvent;
+
+ public RabbitmqCloudEventWriter() {
+ rabbitmqCloudEvent = new RabbitmqCloudEvent();
+ }
+
+ @Override
+ public RabbitmqCloudEvent setEvent(EventFormat format, byte[] value)
throws CloudEventRWException {
+ rabbitmqCloudEvent.setData(new String(value, StandardCharsets.UTF_8));
+ return rabbitmqCloudEvent;
+ }
+
+ @Override
+ public RabbitmqCloudEvent end(CloudEventData data) throws
CloudEventRWException {
+ rabbitmqCloudEvent.setData(new String(data.toBytes(),
StandardCharsets.UTF_8));
+ return rabbitmqCloudEvent;
+ }
+
+ @Override
+ public RabbitmqCloudEvent end() throws CloudEventRWException {
+ rabbitmqCloudEvent.setData("");
+ return rabbitmqCloudEvent;
+ }
+
+ @Override
+ public CloudEventContextWriter withContextAttribute(String name, String
value) throws CloudEventRWException {
+ rabbitmqCloudEvent.getExtensions().put(name, value);
+ return this;
+ }
+
+ @Override
+ public CloudEventWriter<RabbitmqCloudEvent> create(SpecVersion version)
throws CloudEventRWException {
+ rabbitmqCloudEvent.setVersion(version);
+ return this;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolder.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolder.java
new file mode 100644
index 000000000..21e2efd7b
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolder.java
@@ -0,0 +1,60 @@
+/*
+ * 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.storage.rabbitmq.config;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+
+import lombok.Data;
+
+@Data
+@Config(prefix = "eventMesh.server.rabbitmq", path =
"classPath://rabbitmq-client.properties")
+public class ConfigurationHolder {
+
+ @ConfigFiled(field = "host")
+ private String host;
+
+ @ConfigFiled(field = "port")
+ private int port;
+
+ @ConfigFiled(field = "username")
+ private String username;
+
+ @ConfigFiled(field = "passwd")
+ private String passwd;
+
+ @ConfigFiled(field = "virtualHost")
+ private String virtualHost;
+
+ @ConfigFiled(field = "exchangeType")
+ private BuiltinExchangeType exchangeType;
+
+ @ConfigFiled(field = "exchangeName")
+ private String exchangeName;
+
+ @ConfigFiled(field = "routingKey")
+ private String routingKey;
+
+ @ConfigFiled(field = "queueName")
+ private String queueName;
+
+ @ConfigFiled(field = "autoAck")
+ private boolean autoAck;
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumer.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumer.java
new file mode 100644
index 000000000..58b500657
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumer.java
@@ -0,0 +1,143 @@
+/*
+ * 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.storage.rabbitmq.consumer;
+
+import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.consumer.Consumer;
+import org.apache.eventmesh.common.ThreadPoolFactory;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.rabbitmq.client.RabbitmqClient;
+import org.apache.eventmesh.storage.rabbitmq.client.RabbitmqConnectionFactory;
+import org.apache.eventmesh.storage.rabbitmq.config.ConfigurationHolder;
+
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ThreadPoolExecutor;
+
+
+import io.cloudevents.CloudEvent;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Config(field = "configurationHolder")
+
+@Slf4j
+public class RabbitmqConsumer implements Consumer {
+
+
+ private RabbitmqConnectionFactory rabbitmqConnectionFactory = new
RabbitmqConnectionFactory();
+
+ private RabbitmqClient rabbitmqClient;
+
+ private Connection connection;
+
+ private Channel channel;
+
+ private volatile boolean started = false;
+
+ /**
+ * Unified configuration class corresponding to rabbitmq-client.properties
+ */
+ private ConfigurationHolder configurationHolder;
+
+ private final ThreadPoolExecutor executor =
ThreadPoolFactory.createThreadPoolExecutor(
+ Runtime.getRuntime().availableProcessors() * 2,
+ Runtime.getRuntime().availableProcessors() * 2,
+ "EventMesh-Rabbitmq-Consumer");
+
+ private RabbitmqConsumerHandler rabbitmqConsumerHandler;
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !isStarted();
+ }
+
+ @Override
+ public void start() {
+ if (!started) {
+ started = true;
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ if (started) {
+ try {
+ rabbitmqClient.closeConnection(connection);
+ rabbitmqClient.closeChannel(channel);
+ rabbitmqConsumerHandler.stop();
+ } finally {
+ started = false;
+ }
+ }
+ }
+
+ @Override
+ public void init(Properties keyValue) throws Exception {
+ this.rabbitmqClient = new RabbitmqClient(rabbitmqConnectionFactory);
+ this.connection =
rabbitmqClient.getConnection(configurationHolder.getHost(),
configurationHolder.getUsername(),
+ configurationHolder.getPasswd(), configurationHolder.getPort(),
configurationHolder.getVirtualHost());
+ this.channel = rabbitmqConnectionFactory.createChannel(connection);
+ this.rabbitmqConsumerHandler = new RabbitmqConsumerHandler(channel,
configurationHolder);
+ }
+
+ @Override
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+
+ }
+
+ @Override
+ public void subscribe(String topic) {
+ rabbitmqClient.binding(channel, configurationHolder.getExchangeType(),
configurationHolder.getExchangeName(),
+ configurationHolder.getRoutingKey(),
configurationHolder.getQueueName());
+ executor.execute(rabbitmqConsumerHandler);
+ }
+
+ @Override
+ public void unsubscribe(String topic) {
+ try {
+ rabbitmqClient.unbinding(channel,
configurationHolder.getExchangeName(),
+ configurationHolder.getRoutingKey(),
configurationHolder.getQueueName());
+ rabbitmqConsumerHandler.stop();
+ } catch (Exception ex) {
+ log.error("[RabbitmqConsumer] unsubscribe happen exception.", ex);
+ }
+ }
+
+ @Override
+ public void registerEventListener(EventListener listener) {
+ rabbitmqConsumerHandler.setEventListener(listener);
+ }
+
+ public void setRabbitmqConnectionFactory(RabbitmqConnectionFactory
rabbitmqConnectionFactory) {
+ this.rabbitmqConnectionFactory = rabbitmqConnectionFactory;
+ }
+
+ public ConfigurationHolder getClientConfiguration() {
+ return this.configurationHolder;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerHandler.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerHandler.java
new file mode 100644
index 000000000..c52b859a2
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerHandler.java
@@ -0,0 +1,83 @@
+/*
+ * 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.storage.rabbitmq.consumer;
+
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
+import org.apache.eventmesh.storage.rabbitmq.cloudevent.RabbitmqCloudEvent;
+import org.apache.eventmesh.storage.rabbitmq.config.ConfigurationHolder;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.cloudevents.CloudEvent;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.GetResponse;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RabbitmqConsumerHandler implements Runnable {
+
+
+ private final Channel channel;
+ private final ConfigurationHolder configurationHolder;
+ private final AtomicBoolean stop = new AtomicBoolean(false);
+ private EventListener eventListener;
+
+ public RabbitmqConsumerHandler(Channel channel, ConfigurationHolder
configurationHolder) {
+ this.channel = channel;
+ this.configurationHolder = configurationHolder;
+ }
+
+ @Override
+ public void run() {
+ while (!stop.get()) {
+ try {
+ GetResponse response =
channel.basicGet(configurationHolder.getQueueName(),
configurationHolder.isAutoAck());
+ if (response != null) {
+ RabbitmqCloudEvent rabbitmqCloudEvent =
RabbitmqCloudEvent.getFromByteArray(response.getBody());
+ CloudEvent cloudEvent =
rabbitmqCloudEvent.convertToCloudEvent();
+ final EventMeshAsyncConsumeContext consumeContext = new
EventMeshAsyncConsumeContext() {
+ @Override
+ public void commit(EventMeshAction action) {
+ log.info("[RabbitmqConsumerHandler] Rabbitmq
consumer context commit.");
+ }
+ };
+ if (eventListener != null) {
+ eventListener.consume(cloudEvent, consumeContext);
+ }
+ if (!configurationHolder.isAutoAck()) {
+
channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
+ }
+ }
+ } catch (Exception ex) {
+ log.error("[RabbitmqConsumerHandler] thread run happen
exception.", ex);
+ }
+ }
+ }
+
+ public void setEventListener(EventListener eventListener) {
+ this.eventListener = eventListener;
+ }
+
+ public void stop() {
+ stop.set(true);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/exception/RabbitmqStorageException.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/exception/RabbitmqStorageException.java
new file mode 100644
index 000000000..80d210957
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/exception/RabbitmqStorageException.java
@@ -0,0 +1,31 @@
+/*
+ * 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.storage.rabbitmq.exception;
+
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+
+public class RabbitmqStorageException extends StorageConnectorRuntimeException
{
+
+ public RabbitmqStorageException(String message) {
+ super(message);
+ }
+
+ public RabbitmqStorageException(Throwable throwable) {
+ super(throwable);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducer.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducer.java
new file mode 100644
index 000000000..01fc96af4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducer.java
@@ -0,0 +1,167 @@
+/*
+ * 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.storage.rabbitmq.producer;
+
+import org.apache.eventmesh.api.RequestReplyCallback;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+import org.apache.eventmesh.api.producer.Producer;
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.storage.rabbitmq.client.RabbitmqClient;
+import org.apache.eventmesh.storage.rabbitmq.client.RabbitmqConnectionFactory;
+import org.apache.eventmesh.storage.rabbitmq.cloudevent.RabbitmqCloudEvent;
+import
org.apache.eventmesh.storage.rabbitmq.cloudevent.RabbitmqCloudEventWriter;
+import org.apache.eventmesh.storage.rabbitmq.config.ConfigurationHolder;
+import org.apache.eventmesh.storage.rabbitmq.utils.ByteArrayUtils;
+
+import java.util.Optional;
+import java.util.Properties;
+
+import io.cloudevents.CloudEvent;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Config(field = "configurationHolder")
+@Slf4j
+public class RabbitmqProducer implements Producer {
+
+ private RabbitmqConnectionFactory rabbitmqConnectionFactory = new
RabbitmqConnectionFactory();
+
+ private RabbitmqClient rabbitmqClient;
+
+ private Connection connection;
+
+ private Channel channel;
+
+ private volatile boolean started = false;
+
+ /**
+ * Unified configuration class corresponding to rabbitmq-client.properties
+ */
+ private ConfigurationHolder configurationHolder;
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !isStarted();
+ }
+
+ @Override
+ public void start() {
+ if (!started) {
+ started = true;
+ }
+ }
+
+ @Override
+ public void shutdown() {
+ if (started) {
+ try {
+ rabbitmqClient.closeConnection(connection);
+ rabbitmqClient.closeChannel(channel);
+ } finally {
+ started = false;
+ }
+ }
+ }
+
+ @Override
+ public void init(Properties properties) throws Exception {
+ this.rabbitmqClient = new RabbitmqClient(rabbitmqConnectionFactory);
+ this.connection =
rabbitmqClient.getConnection(configurationHolder.getHost(),
configurationHolder.getUsername(),
+ configurationHolder.getPasswd(), configurationHolder.getPort(),
configurationHolder.getVirtualHost());
+ this.channel = rabbitmqConnectionFactory.createChannel(connection);
+ }
+
+ @Override
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ try {
+ RabbitmqCloudEventWriter writer = new RabbitmqCloudEventWriter();
+ RabbitmqCloudEvent rabbitmqCloudEvent =
writer.writeBinary(cloudEvent);
+ byte[] data = RabbitmqCloudEvent.toByteArray(rabbitmqCloudEvent);
+ if (data != null) {
+ rabbitmqClient.publish(channel,
configurationHolder.getExchangeName(), configurationHolder.getRoutingKey(),
data);
+
+ SendResult sendResult = new SendResult();
+ sendResult.setTopic(cloudEvent.getSubject());
+ sendResult.setMessageId(cloudEvent.getId());
+ sendCallback.onSuccess(sendResult);
+ }
+ } catch (Exception ex) {
+ log.error("[RabbitmqProducer] publish happen exception.", ex);
+ sendCallback.onException(
+ OnExceptionContext.builder()
+ .topic(cloudEvent.getSubject())
+ .messageId(cloudEvent.getId())
+ .exception(new StorageConnectorRuntimeException(ex))
+ .build()
+ );
+ }
+ }
+
+ @Override
+ public void sendOneway(CloudEvent cloudEvent) {
+ try {
+ Optional<byte[]> optionalBytes =
ByteArrayUtils.objectToBytes(cloudEvent);
+ if (optionalBytes.isPresent()) {
+ byte[] data = optionalBytes.get();
+ rabbitmqClient.publish(channel,
configurationHolder.getExchangeName(),
+ configurationHolder.getRoutingKey(), data);
+ }
+ } catch (Exception ex) {
+ log.error("[RabbitmqProducer] sendOneway happen exception.", ex);
+ }
+ }
+
+ @Override
+ public void request(CloudEvent cloudEvent, RequestReplyCallback
rrCallback, long timeout) throws Exception {
+
+ }
+
+ @Override
+ public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback)
throws Exception {
+ return false;
+ }
+
+ @Override
+ public void checkTopicExist(String topic) throws Exception {
+
+ }
+
+ @Override
+ public void setExtFields() {
+
+ }
+
+ public void setRabbitmqConnectionFactory(RabbitmqConnectionFactory
rabbitmqConnectionFactory) {
+ this.rabbitmqConnectionFactory = rabbitmqConnectionFactory;
+ }
+
+ public ConfigurationHolder getClientConfiguration() {
+ return this.configurationHolder;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/storage/RabbitmqStorageResourceService.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/storage/RabbitmqStorageResourceService.java
new file mode 100644
index 000000000..260abf4b2
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/storage/RabbitmqStorageResourceService.java
@@ -0,0 +1,33 @@
+/*
+ * 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.storage.rabbitmq.storage;
+
+import org.apache.eventmesh.api.storage.StorageResourceService;
+
+public class RabbitmqStorageResourceService implements StorageResourceService {
+
+ @Override
+ public void init() throws Exception {
+
+ }
+
+ @Override
+ public void release() throws Exception {
+
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/utils/ByteArrayUtils.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/utils/ByteArrayUtils.java
new file mode 100644
index 000000000..33db09c57
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/java/org/apache/eventmesh/storage/rabbitmq/utils/ByteArrayUtils.java
@@ -0,0 +1,42 @@
+/*
+ * 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.storage.rabbitmq.utils;
+
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.utils.JsonUtils;
+
+import java.io.IOException;
+import java.util.Optional;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+
+@SuppressWarnings("all")
+public class ByteArrayUtils {
+
+ public static <T> Optional<byte[]> objectToBytes(T obj) throws IOException
{
+ String s = JsonUtils.toJSONString(obj);
+ byte[] bytes = s.getBytes();
+ return Optional.ofNullable(bytes);
+ }
+
+ public static <T> Optional<T> bytesToObject(byte[] bytes) throws
IOException, ClassNotFoundException {
+ T t = JsonUtils.parseTypeReferenceObject(new String(bytes,
Constants.DEFAULT_CHARSET), new TypeReference<T>() {
+ });
+ return Optional.ofNullable(t);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..1599d924f
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
@@ -0,0 +1,16 @@
+# 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.
+
+rabbitmq=org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumer
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
new file mode 100644
index 000000000..80c89f0f1
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
@@ -0,0 +1,16 @@
+# 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.
+
+rabbitmq=org.apache.eventmesh.storage.rabbitmq.producer.RabbitmqProducer
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
new file mode 100644
index 000000000..1324e455f
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
@@ -0,0 +1,16 @@
+# 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.
+
+rabbitmq=org.apache.eventmesh.connector.rabbitmq.storage.RabbitmqStorageResourceService
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/rabbitmq-client.properties
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/rabbitmq-client.properties
new file mode 100644
index 000000000..db9f3a5fa
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/main/resources/rabbitmq-client.properties
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+####################### rabbitmq server ##################
+eventMesh.server.rabbitmq.host=
+eventMesh.server.rabbitmq.port=
+eventMesh.server.rabbitmq.username=
+eventMesh.server.rabbitmq.passwd=
+eventMesh.server.rabbitmq.virtualHost=
+
+####################### rabbitmq queue setting ##################
+# DIRECT, FANOUT, TOPIC, HEADERS
+eventMesh.server.rabbitmq.exchangeType=TOPIC
+eventMesh.server.rabbitmq.exchangeName=
+eventMesh.server.rabbitmq.routingKey=
+eventMesh.server.rabbitmq.queueName=
+eventMesh.server.rabbitmq.autoAck=true
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqMockConnectionFactory.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqMockConnectionFactory.java
new file mode 100644
index 000000000..ffda7ca44
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqMockConnectionFactory.java
@@ -0,0 +1,59 @@
+/*
+ * 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.storage.rabbitmq;
+
+import org.apache.eventmesh.storage.rabbitmq.client.RabbitmqConnectionFactory;
+
+import com.github.fridujo.rabbitmq.mock.MockConnectionFactory;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+
+public class RabbitmqMockConnectionFactory extends RabbitmqConnectionFactory {
+
+ private final ConnectionFactory myConnectionFactory;
+
+ private final Connection myConnection;
+
+ private final Channel myChannel;
+
+ public RabbitmqMockConnectionFactory() throws Exception {
+ this.myConnectionFactory = new MockConnectionFactory();
+ this.myConnectionFactory.setHost("127.0.0.1");
+ this.myConnectionFactory.setPort(5672);
+ this.myConnectionFactory.setUsername("root");
+ this.myConnectionFactory.setPassword("123456");
+ this.myConnection = this.myConnectionFactory.newConnection();
+ this.myChannel = myConnection.createChannel();
+ }
+
+ @Override
+ public ConnectionFactory createConnectionFactory() {
+ return this.myConnectionFactory;
+ }
+
+ @Override
+ public Connection createConnection(ConnectionFactory connectionFactory) {
+ return this.myConnection;
+ }
+
+ @Override
+ public Channel createChannel(Connection connection) {
+ return this.myChannel;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqServer.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqServer.java
new file mode 100644
index 000000000..175f56bc6
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/RabbitmqServer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.storage.rabbitmq;
+
+import org.apache.eventmesh.api.factory.StoragePluginFactory;
+import org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumer;
+import org.apache.eventmesh.storage.rabbitmq.producer.RabbitmqProducer;
+
+import java.util.Properties;
+
+import org.junit.After;
+import org.junit.Before;
+
+public class RabbitmqServer {
+
+ protected RabbitmqConsumer rabbitmqConsumer;
+ protected RabbitmqProducer rabbitmqProducer;
+
+ @Before
+ public void setup() throws Exception {
+ RabbitmqMockConnectionFactory rabbitmqMockConnectionFactory = new
RabbitmqMockConnectionFactory();
+
+ rabbitmqConsumer =
+ (RabbitmqConsumer)
StoragePluginFactory.getMeshMQPushConsumer("rabbitmq");
+
rabbitmqConsumer.setRabbitmqConnectionFactory(rabbitmqMockConnectionFactory);
+ rabbitmqConsumer.init(new Properties());
+ rabbitmqConsumer.start();
+
+ rabbitmqProducer = (RabbitmqProducer)
StoragePluginFactory.getMeshMQProducer("rabbitmq");
+
rabbitmqProducer.setRabbitmqConnectionFactory(rabbitmqMockConnectionFactory);
+ rabbitmqProducer.init(new Properties());
+ rabbitmqProducer.start();
+ }
+
+ @After
+ public void shutdown() {
+ rabbitmqConsumer.shutdown();
+ rabbitmqProducer.shutdown();
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventTest.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventTest.java
new file mode 100644
index 000000000..e6cbcaae5
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/cloudevent/RabbitmqCloudEventTest.java
@@ -0,0 +1,70 @@
+/*
+ * 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.storage.rabbitmq.cloudevent;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class RabbitmqCloudEventTest {
+
+ private CloudEvent cloudEvent;
+
+ @Before
+ public void before() {
+ cloudEvent = CloudEventBuilder.v1()
+ .withId("1")
+ .withTime(OffsetDateTime.now())
+ .withSource(URI.create("testsource"))
+ .withSubject("topic")
+ .withType(String.class.getCanonicalName())
+ .withDataContentType("text/plain")
+ .withData("data".getBytes(StandardCharsets.UTF_8))
+ .build();
+ }
+
+ @Test
+ public void toByteArray() throws Exception {
+ RabbitmqCloudEventWriter writer = new RabbitmqCloudEventWriter();
+ RabbitmqCloudEvent rabbitmqCloudEvent = writer.writeBinary(cloudEvent);
+ Assert.assertEquals(cloudEvent.getSubject(), "topic");
+
+ byte[] data = RabbitmqCloudEvent.toByteArray(rabbitmqCloudEvent);
+ Assert.assertNotNull(data);
+ }
+
+ @Test
+ public void getFromByteArray() throws Exception {
+ RabbitmqCloudEventWriter writer = new RabbitmqCloudEventWriter();
+ RabbitmqCloudEvent rabbitmqCloudEvent = writer.writeBinary(cloudEvent);
+ Assert.assertEquals(cloudEvent.getSubject(), "topic");
+
+ byte[] data = RabbitmqCloudEvent.toByteArray(rabbitmqCloudEvent);
+ Assert.assertNotNull(data);
+
+ RabbitmqCloudEvent event = RabbitmqCloudEvent.getFromByteArray(data);
+ Assert.assertEquals(event.getExtensions().get("subject"), "topic");
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolderTest.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolderTest.java
new file mode 100644
index 000000000..6d98e4d1b
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/config/ConfigurationHolderTest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.storage.rabbitmq.config;
+
+import org.apache.eventmesh.api.factory.StoragePluginFactory;
+import org.apache.eventmesh.storage.rabbitmq.consumer.RabbitmqConsumer;
+import org.apache.eventmesh.storage.rabbitmq.producer.RabbitmqProducer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.rabbitmq.client.BuiltinExchangeType;
+
+public class ConfigurationHolderTest {
+
+ @Test
+ public void getConfigWhenRabbitmqConsumerInit() {
+ RabbitmqConsumer consumer =
+ (RabbitmqConsumer)
StoragePluginFactory.getMeshMQPushConsumer("rabbitmq");
+
+ ConfigurationHolder config = consumer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ @Test
+ public void getConfigWhenRabbitmqProducerInit() {
+ RabbitmqProducer producer =
+ (RabbitmqProducer)
StoragePluginFactory.getMeshMQProducer("rabbitmq");
+
+ ConfigurationHolder config = producer.getClientConfiguration();
+ assertConfig(config);
+ }
+
+ private void assertConfig(ConfigurationHolder config) {
+ Assert.assertEquals(config.getHost(), "127.0.0.1");
+ Assert.assertEquals(config.getPort(), 5672);
+ Assert.assertEquals(config.getUsername(), "username-success!!!");
+ Assert.assertEquals(config.getPasswd(), "passwd-success!!!");
+ Assert.assertEquals(config.getVirtualHost(), "virtualHost-success!!!");
+
+ Assert.assertEquals(config.getExchangeType(),
BuiltinExchangeType.TOPIC);
+ Assert.assertEquals(config.getExchangeName(),
"exchangeName-success!!!");
+ Assert.assertEquals(config.getRoutingKey(), "routingKey-success!!!");
+ Assert.assertEquals(config.getQueueName(), "queueName-success!!!");
+ Assert.assertTrue(config.isAutoAck());
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerTest.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerTest.java
new file mode 100644
index 000000000..f6a933634
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/consumer/RabbitmqConsumerTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.storage.rabbitmq.consumer;
+
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.utils.ThreadUtils;
+import org.apache.eventmesh.storage.rabbitmq.RabbitmqServer;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class RabbitmqConsumerTest extends RabbitmqServer {
+
+ @Test
+ public void isStarted() {
+ Assert.assertTrue(rabbitmqConsumer.isStarted());
+ }
+
+ @Test
+ public void isClosed() {
+ Assert.assertFalse(rabbitmqConsumer.isClosed());
+ }
+
+ @Test
+ public void subscribe() throws Exception {
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+
+ rabbitmqConsumer.registerEventListener((cloudEvent, context) -> {
+ downLatch.countDown();
+ context.commit(EventMeshAction.CommitMessage);
+ Assert.assertEquals(cloudEvent.getSubject(), "topic");
+ });
+
+ rabbitmqConsumer.subscribe("topic");
+
+ ThreadUtils.sleep(1, TimeUnit.SECONDS);
+ for (int i = 0; i < expectedCount; i++) {
+ CloudEvent cloudEvent = CloudEventBuilder.v1()
+ .withId(String.valueOf(i))
+ .withTime(OffsetDateTime.now())
+ .withSource(URI.create("testsource"))
+ .withSubject("topic")
+ .withType(String.class.getCanonicalName())
+ .withDataContentType("text/plain")
+ .withData("data".getBytes(StandardCharsets.UTF_8))
+ .build();
+
+ rabbitmqProducer.publish(cloudEvent, new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ Assert.assertEquals(cloudEvent.getId(),
sendResult.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
sendResult.getTopic());
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+ Assert.assertEquals(cloudEvent.getId(),
context.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
context.getTopic());
+ }
+ });
+ }
+
+ Assert.assertTrue(downLatch.await(5, TimeUnit.MINUTES));
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducerTest.java
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducerTest.java
new file mode 100644
index 000000000..313669bda
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/java/org/apache/eventmesh/storage/rabbitmq/producer/RabbitmqProducerTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.storage.rabbitmq.producer;
+
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.common.utils.ThreadUtils;
+import org.apache.eventmesh.storage.rabbitmq.RabbitmqServer;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class RabbitmqProducerTest extends RabbitmqServer {
+
+ @Test
+ public void isStarted() {
+ Assert.assertTrue(rabbitmqProducer.isStarted());
+ }
+
+ @Test
+ public void isClosed() {
+ Assert.assertFalse(rabbitmqProducer.isClosed());
+ }
+
+ @Test
+ public void publish() throws Exception {
+ final int expectedCount = 5;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+
+ rabbitmqConsumer.registerEventListener((cloudEvent, context) -> {
+ downLatch.countDown();
+ context.commit(EventMeshAction.CommitMessage);
+ Assert.assertEquals(cloudEvent.getSubject(), "topic");
+ });
+
+ rabbitmqConsumer.subscribe("topic");
+
+ ThreadUtils.sleep(1, TimeUnit.SECONDS);
+ for (int i = 0; i < expectedCount; i++) {
+ CloudEvent cloudEvent = CloudEventBuilder.v1()
+ .withId(String.valueOf(i))
+ .withTime(OffsetDateTime.now())
+ .withSource(URI.create("testsource"))
+ .withSubject("topic")
+ .withType(String.class.getCanonicalName())
+ .withDataContentType("text/plain")
+ .withData("data".getBytes(StandardCharsets.UTF_8))
+ .build();
+
+ rabbitmqProducer.publish(cloudEvent, new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ Assert.assertEquals(cloudEvent.getId(),
sendResult.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
sendResult.getTopic());
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+ Assert.assertEquals(cloudEvent.getId(),
context.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
context.getTopic());
+ }
+ });
+ }
+
+ Assert.assertTrue(downLatch.await(5, TimeUnit.MINUTES));
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/resources/rabbitmq-client.properties
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/resources/rabbitmq-client.properties
new file mode 100644
index 000000000..082fbe5b4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-rabbitmq/src/test/resources/rabbitmq-client.properties
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+####################### rabbitmq server ##################
+eventMesh.server.rabbitmq.host=127.0.0.1
+eventMesh.server.rabbitmq.port=5672
+eventMesh.server.rabbitmq.username=username-success!!!
+eventMesh.server.rabbitmq.passwd=passwd-success!!!
+eventMesh.server.rabbitmq.virtualHost=virtualHost-success!!!
+
+####################### rabbitmq queue setting ##################
+# DIRECT, FANOUT, TOPIC, HEADERS
+eventMesh.server.rabbitmq.exchangeType=TOPIC
+eventMesh.server.rabbitmq.exchangeName=exchangeName-success!!!
+eventMesh.server.rabbitmq.routingKey=routingKey-success!!!
+eventMesh.server.rabbitmq.queueName=queueName-success!!!
+eventMesh.server.rabbitmq.autoAck=true
\ No newline at end of file
diff --git a/settings.gradle b/settings.gradle
index 8456e9430..491c6a47e 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -28,6 +28,7 @@ include 'eventmesh-storage:eventmesh-storage-standalone'
include 'eventmesh-storage:eventmesh-storage-kafka'
include 'eventmesh-storage:eventmesh-storage-redis'
include 'eventmesh-storage:eventmesh-storage-rocketmq'
+include 'eventmesh-storage:eventmesh-storage-rabbitmq'
include 'eventmesh-connector-plugin:eventmesh-connector-api'
include 'eventmesh-connector-plugin:eventmesh-connector-knative'
include 'eventmesh-connector-plugin:eventmesh-connector-pravega'
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
