This is an automated email from the ASF dual-hosted git repository.
chenguangsheng 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 670867656 [ISSUE #3370] Add eventmesh-storage-redis module
new d5c97e819 Merge pull request #3371 from xwm1992/add-storage-redis
670867656 is described below
commit 670867656102cf902635056b714f0bbfbc80e41d
Author: xwm1992 <[email protected]>
AuthorDate: Wed Mar 8 20:11:45 2023 +0800
[ISSUE #3370] Add eventmesh-storage-redis module
---
.../eventmesh-storage-redis/build.gradle | 56 +++++++++
.../eventmesh-storage-redis/gradle.properties | 18 +++
.../storage/redis/client/RedissonClient.java | 114 +++++++++++++++++
.../storage/redis/cloudevent/CloudEventCodec.java | 60 +++++++++
.../storage/redis/config/RedisProperties.java | 66 ++++++++++
.../storage/redis/consumer/RedisConsumer.java | 131 +++++++++++++++++++
.../storage/redis/producer/RedisProducer.java | 139 +++++++++++++++++++++
.../redis/storage/RedisStorageResourceService.java | 33 +++++
.../org.apache.eventmesh.api.consumer.Consumer | 16 +++
.../org.apache.eventmesh.api.producer.Producer | 16 +++
...he.eventmesh.api.storage.StorageResourceService | 16 +++
.../src/main/resources/redis-client.properties | 17 +++
.../storage/redis/AbstractRedisServer.java | 41 ++++++
.../storage/redis/client/RedissonClientTest.java | 31 +++++
.../storage/redis/config/RedisPropertiesTest.java | 47 +++++++
.../storage/redis/connector/UnitTest.java | 110 ++++++++++++++++
.../storage/redis/consumer/RedisConsumerTest.java | 91 ++++++++++++++
.../storage/redis/producer/RedisProducerTest.java | 110 ++++++++++++++++
.../src/test/resources/redis-client.properties | 21 ++++
19 files changed, 1133 insertions(+)
diff --git a/eventmesh-storage/eventmesh-storage-redis/build.gradle
b/eventmesh-storage/eventmesh-storage-redis/build.gradle
new file mode 100644
index 000000000..b681d7cff
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-redis/build.gradle
@@ -0,0 +1,56 @@
+/*
+ * 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: 'org.slf4j', module: 'slf4j-simple'
+}
+
+dependencies {
+ implementation project(":eventmesh-common")
+ implementation project(":eventmesh-storage:eventmesh-storage-api")
+
+ // redisson
+ implementation('org.redisson:redisson:3.17.3') {
+ exclude group: 'io.netty', module: 'netty-common'
+ exclude group: 'io.netty', module: 'netty-buffer'
+ exclude group: 'io.netty', module: 'netty-codec'
+ exclude group: 'io.netty', module: 'netty-transport'
+ exclude group: 'io.netty', module: 'netty-resolver'
+ exclude group: 'io.netty', module: 'netty-resolver-dns'
+ exclude group: 'io.netty', module: 'netty-handler'
+ }
+
+ // netty
+ implementation "io.netty:netty-all"
+
+ // auxiliary serialize
+ api 'io.cloudevents:cloudevents-json-jackson'
+
+ // test dependencies
+ testImplementation 'ai.grakn:redis-mock:0.1.6'
+ testImplementation "org.mockito:mockito-core"
+
+ compileOnly 'org.projectlombok:lombok'
+ annotationProcessor 'org.projectlombok:lombok'
+
+ testCompileOnly 'org.projectlombok:lombok'
+ testAnnotationProcessor 'org.projectlombok:lombok'
+}
+
+test {
+ systemProperty "io.netty.tryUnsafe", "false"
+}
diff --git a/eventmesh-storage/eventmesh-storage-redis/gradle.properties
b/eventmesh-storage/eventmesh-storage-redis/gradle.properties
new file mode 100644
index 000000000..899c915a5
--- /dev/null
+++ b/eventmesh-storage/eventmesh-storage-redis/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=redis
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/client/RedissonClient.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/client/RedissonClient.java
new file mode 100644
index 000000000..43f4181e4
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/client/RedissonClient.java
@@ -0,0 +1,114 @@
+/*
+ * 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.redis.client;
+
+import org.apache.eventmesh.api.exception.StorageConnectorRuntimeException;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.common.config.ConfigService;
+import org.apache.eventmesh.storage.redis.cloudevent.CloudEventCodec;
+import org.apache.eventmesh.storage.redis.config.RedisProperties;
+
+import java.util.Arrays;
+
+import org.redisson.Redisson;
+import org.redisson.config.Config;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+/**
+ * Within EventMesh's JVM, there is no multi-connector server, and redisson
itself is pooled management, so a single instance is fine, and it can save
+ * resources and improve performance.
+ */
+public final class RedissonClient {
+
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ public static final Redisson INSTANCE;
+
+ static {
+ OBJECT_MAPPER.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS,
false);
+
+ INSTANCE = create();
+
+ Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+ try {
+ INSTANCE.shutdown();
+ } catch (Exception ignore) {
+ //
+ }
+ }));
+ }
+
+ public static Redisson create() {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties properties =
configService.buildConfigInstance(RedisProperties.class);
+
+ return create(properties);
+ }
+
+ private static Redisson create(RedisProperties properties) {
+ RedisProperties.ServerType serverType;
+ try {
+ serverType = properties.getServerType();
+ } catch (IllegalArgumentException ie) {
+ final String message = "Invalid Redis server type: " +
properties.getServerType()
+ + ", supported values are: "
+ + Arrays.toString(RedisProperties.ServerType.values());
+ throw new StorageConnectorRuntimeException(message, ie);
+ }
+
+ String serverAddress = properties.getServerAddress();
+ String serverPassword = properties.getServerPassword();
+ String masterName = properties.getServerMasterName();
+
+ Config config =
OBJECT_MAPPER.convertValue(properties.getRedissonProperties(), Config.class);
+
+ if (config == null) {
+ config = new Config();
+ }
+
+ config.setCodec(CloudEventCodec.INSTANCE);
+
+ switch (serverType) {
+ case SINGLE:
+ config.useSingleServer()
+ .setAddress(serverAddress)
+ .setPassword(serverPassword);
+ break;
+ case CLUSTER:
+ config.useClusterServers()
+ .addNodeAddress(serverAddress.split(Constants.COMMA))
+ .setPassword(serverPassword);
+ break;
+ case SENTINEL:
+ config.useSentinelServers()
+ .setMasterName(masterName)
+ .addSentinelAddress(serverAddress)
+ .setPassword(serverPassword);
+ break;
+ default:
+ final String message = "Invalid Redis server type: " +
properties.getServerType()
+ + ", supported values are: "
+ + Arrays.toString(RedisProperties.ServerType.values());
+ throw new StorageConnectorRuntimeException(message);
+ }
+
+ return (Redisson) Redisson.create(config);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
new file mode 100644
index 000000000..932488d8c
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.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.redis.cloudevent;
+
+import org.redisson.client.codec.BaseCodec;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.jackson.JsonFormat;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+
+
+public class CloudEventCodec extends BaseCodec {
+
+ public static final CloudEventCodec INSTANCE = new CloudEventCodec();
+
+ private static final JsonFormat jsonFormat = new JsonFormat(false, true);
+
+ private static final Encoder encoder = in -> {
+ ByteBuf out = ByteBufAllocator.DEFAULT.buffer();
+ if (in instanceof CloudEvent) {
+ out.writeBytes(jsonFormat.serialize((CloudEvent) in));
+ return out;
+ }
+ throw new IllegalStateException("Illegal object type: " +
in.getClass().getSimpleName());
+ };
+
+ private static final Decoder<Object> decoder = (buf, state) -> {
+ final byte[] bytes = new byte[buf.readableBytes()];
+ buf.getBytes(buf.readerIndex(), bytes);
+ return jsonFormat.deserialize(bytes);
+ };
+
+ @Override
+ public Decoder<Object> getValueDecoder() {
+ return decoder;
+ }
+
+ @Override
+ public Encoder getValueEncoder() {
+ return encoder;
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/config/RedisProperties.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/config/RedisProperties.java
new file mode 100644
index 000000000..9ac97b1db
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/config/RedisProperties.java
@@ -0,0 +1,66 @@
+/*
+ * 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.redis.config;
+
+import org.apache.eventmesh.common.config.Config;
+import org.apache.eventmesh.common.config.ConfigFiled;
+
+import java.util.Properties;
+
+import lombok.Data;
+
+@Data
+@Config(prefix = "eventMesh.server.redis", path =
"classPath://redis-client.properties")
+public class RedisProperties {
+
+ /**
+ * The redis server configuration to be used.
+ */
+ @ConfigFiled(field = "serverType")
+ private ServerType serverType = ServerType.SINGLE;
+
+ /**
+ * The master server name used by Redis Sentinel servers and master change
monitoring task.
+ */
+ @ConfigFiled(field = "serverMasterName")
+ private String serverMasterName = "master";
+
+ /**
+ * The address of the redis server following format --
host1:port1,host2:port2,……
+ */
+ @ConfigFiled(field = "serverAddress")
+ private String serverAddress;
+
+ /**
+ * The password for redis authentication.
+ */
+ @ConfigFiled(field = "serverPassword")
+ private String serverPassword;
+
+ /**
+ * The redisson options, redisson properties prefix is
`eventMesh.server.redis.redisson`
+ */
+ @ConfigFiled(field = "redisson")
+ private Properties redissonProperties;
+
+ public enum ServerType {
+ SINGLE,
+ CLUSTER,
+ SENTINEL
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumer.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumer.java
new file mode 100644
index 000000000..5d689fe96
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumer.java
@@ -0,0 +1,131 @@
+/*
+ * 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.redis.consumer;
+
+import org.apache.eventmesh.api.AbstractContext;
+import org.apache.eventmesh.api.EventListener;
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.api.EventMeshAsyncConsumeContext;
+import org.apache.eventmesh.api.consumer.Consumer;
+import org.apache.eventmesh.storage.redis.client.RedissonClient;
+
+import java.util.List;
+import java.util.Properties;
+
+import org.redisson.Redisson;
+import org.redisson.api.listener.MessageListener;
+
+import io.cloudevents.CloudEvent;
+
+import com.google.common.base.Preconditions;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class RedisConsumer implements Consumer {
+
+
+ private Redisson redisson;
+
+ private EventMeshMessageListener messageListener;
+
+ private volatile boolean started = false;
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !isStarted();
+ }
+
+ @Override
+ public synchronized void start() {
+ if (!started) {
+ started = true;
+ }
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ if (started) {
+ redisson = null;
+ messageListener = null;
+ started = false;
+ }
+ }
+
+ @Override
+ public void init(Properties keyValue) {
+ // Currently, 'keyValue' does not pass useful configuration
information.
+ redisson = RedissonClient.INSTANCE;
+ }
+
+ @Override
+ public void updateOffset(List<CloudEvent> cloudEvents, AbstractContext
context) {
+
+ }
+
+ @Override
+ public void subscribe(String topic) {
+ Preconditions.checkNotNull(topic);
+ Preconditions.checkNotNull(messageListener);
+
+ redisson.getTopic(topic).addListenerAsync(CloudEvent.class,
messageListener);
+ }
+
+ @Override
+ public void unsubscribe(String topic) {
+ Preconditions.checkNotNull(topic);
+ Preconditions.checkNotNull(messageListener);
+
+ redisson.getTopic(topic).removeListenerAsync(messageListener);
+ }
+
+ @Override
+ public void registerEventListener(EventListener listener) {
+ Preconditions.checkNotNull(listener);
+
+ messageListener = new EventMeshMessageListener(listener);
+ }
+
+ static class EventMeshMessageListener implements
MessageListener<CloudEvent> {
+
+ private final EventListener listener;
+
+ EventMeshMessageListener(EventListener listener) {
+ this.listener = listener;
+ }
+
+ @Override
+ public void onMessage(CharSequence channel, CloudEvent msg) {
+
+ final EventMeshAsyncConsumeContext consumeContext = new
EventMeshAsyncConsumeContext() {
+ @Override
+ public void commit(EventMeshAction action) {
+ log.info("channel: {} consumer event: {} finish action:
{}",
+ channel, msg.getId(), action);
+ }
+ };
+
+ listener.consume(msg, consumeContext);
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/producer/RedisProducer.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/producer/RedisProducer.java
new file mode 100644
index 000000000..329307f41
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/producer/RedisProducer.java
@@ -0,0 +1,139 @@
+/*
+ * 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.redis.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.storage.redis.client.RedissonClient;
+
+import java.util.Properties;
+
+import org.redisson.Redisson;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+
+import com.google.common.base.Preconditions;
+
+public class RedisProducer implements Producer {
+
+ private Redisson redisson;
+
+ private volatile boolean started = false;
+
+ @Override
+ public boolean isStarted() {
+ return started;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !isStarted();
+ }
+
+ @Override
+ public synchronized void start() {
+ if (!started) {
+ started = true;
+ }
+ }
+
+ @Override
+ public synchronized void shutdown() {
+ if (started) {
+ try {
+ redisson = null;
+ } finally {
+ started = false;
+ }
+ }
+ }
+
+ @Override
+ public void init(Properties properties) {
+ // Currently, 'properties' does not pass useful configuration
information.
+ redisson = RedissonClient.INSTANCE;
+ }
+
+ @Override
+ public void publish(CloudEvent cloudEvent, SendCallback sendCallback) {
+ Preconditions.checkNotNull(cloudEvent);
+ Preconditions.checkNotNull(sendCallback);
+
+ try {
+ RTopic topic = redisson.getTopic(cloudEvent.getSubject());
+
+ topic.publishAsync(cloudEvent).whenCompleteAsync((stage,
throwable) -> {
+ if (throwable != null) {
+ sendCallback.onException(
+ OnExceptionContext.builder()
+ .topic(cloudEvent.getSubject())
+ .messageId(cloudEvent.getId())
+ .exception(new
StorageConnectorRuntimeException(throwable))
+ .build()
+ );
+ } else {
+ SendResult sendResult = new SendResult();
+ sendResult.setTopic(cloudEvent.getSubject());
+ sendResult.setMessageId(cloudEvent.getId());
+ sendCallback.onSuccess(sendResult);
+ }
+ });
+ } catch (Exception e) {
+ sendCallback.onException(
+ OnExceptionContext.builder()
+ .topic(cloudEvent.getSubject())
+ .messageId(cloudEvent.getId())
+ .exception(new StorageConnectorRuntimeException(e))
+ .build()
+ );
+ }
+ }
+
+ @Override
+ public void sendOneway(CloudEvent cloudEvent) {
+ Preconditions.checkNotNull(cloudEvent);
+
+ RTopic topic = redisson.getTopic(cloudEvent.getSubject());
+ topic.publish(cloudEvent);
+ }
+
+ @Override
+ public void request(CloudEvent cloudEvent, RequestReplyCallback
rrCallback, long timeout) {
+ throw new StorageConnectorRuntimeException("Request is not supported");
+ }
+
+ @Override
+ public boolean reply(CloudEvent cloudEvent, SendCallback sendCallback) {
+ throw new StorageConnectorRuntimeException("Reply is not supported");
+ }
+
+ @Override
+ public void checkTopicExist(String topic) {
+ // Because redis has the feature of creating topics when used, there
is no need to check existence.
+ }
+
+ @Override
+ public void setExtFields() {
+
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/storage/RedisStorageResourceService.java
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/storage/RedisStorageResourceService.java
new file mode 100644
index 000000000..1846a194d
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/storage/RedisStorageResourceService.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.redis.storage;
+
+import org.apache.eventmesh.api.storage.StorageResourceService;
+
+public class RedisStorageResourceService implements StorageResourceService {
+
+ @Override
+ public void init() throws Exception {
+
+ }
+
+ @Override
+ public void release() throws Exception {
+
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
b/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.consumer.Consumer
new file mode 100644
index 000000000..695ea9aba
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/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.
+
+redis=org.apache.eventmesh.storage.redis.consumer.RedisConsumer
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
b/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.producer.Producer
new file mode 100644
index 000000000..553bf31ce
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/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.
+
+redis=org.apache.eventmesh.storage.redis.producer.RedisProducer
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
b/eventmesh-storage/eventmesh-storage-redis/src/main/resources/META-INF/eventmesh/org.apache.eventmesh.api.storage.StorageResourceService
new file mode 100644
index 000000000..9dd64fbb7
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/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.
+
+redis=org.apache.eventmesh.storage.redis.storage.RedisStorageResourceService
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/main/resources/redis-client.properties
b/eventmesh-storage/eventmesh-storage-redis/src/main/resources/redis-client.properties
new file mode 100644
index 000000000..622297217
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/main/resources/redis-client.properties
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/AbstractRedisServer.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/AbstractRedisServer.java
new file mode 100644
index 000000000..a71eeb6f9
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/AbstractRedisServer.java
@@ -0,0 +1,41 @@
+/*
+ * 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.redis;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+
+import ai.grakn.redismock.RedisServer;
+
+public abstract class AbstractRedisServer {
+
+ private static RedisServer redisServer;
+
+ @BeforeClass
+ public static void setupRedisServer() throws Exception {
+ redisServer = RedisServer.newRedisServer(6379);
+ redisServer.start();
+ }
+
+ @AfterClass
+ public static void shutdownRedisServer() {
+ if (redisServer != null) {
+ redisServer.stop();
+ }
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/client/RedissonClientTest.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/client/RedissonClientTest.java
new file mode 100644
index 000000000..fe18c76c8
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/client/RedissonClientTest.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.redis.client;
+
+import org.apache.eventmesh.storage.redis.AbstractRedisServer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RedissonClientTest extends AbstractRedisServer {
+
+ @Test
+ public void testInstance() {
+ Assert.assertNotNull(RedissonClient.INSTANCE);
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/config/RedisPropertiesTest.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/config/RedisPropertiesTest.java
new file mode 100644
index 000000000..ad4e97785
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/config/RedisPropertiesTest.java
@@ -0,0 +1,47 @@
+/*
+ * 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.redis.config;
+
+import org.apache.eventmesh.common.config.ConfigService;
+
+import java.util.Properties;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RedisPropertiesTest {
+
+ @Test
+ public void getRedisProperties() {
+ ConfigService configService = ConfigService.getInstance();
+ RedisProperties config =
configService.buildConfigInstance(RedisProperties.class);
+ assertConfig(config);
+ }
+
+ private void assertConfig(RedisProperties config) {
+ Assert.assertEquals(config.getServerAddress(),
"redis://127.0.0.1:6379");
+ Assert.assertEquals(config.getServerType(),
RedisProperties.ServerType.SINGLE);
+ Assert.assertEquals(config.getServerMasterName(),
"serverMasterName-success!!!");
+
+ Properties properties = new Properties();
+ properties.put("threads", "2");
+ properties.put("nettyThreads", "2");
+ Properties redissonProperties = config.getRedissonProperties();
+ Assert.assertEquals(redissonProperties, properties);
+ }
+}
\ No newline at end of file
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/connector/UnitTest.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/connector/UnitTest.java
new file mode 100644
index 000000000..d06990cd2
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/connector/UnitTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.redis.connector;
+
+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.storage.redis.AbstractRedisServer;
+import org.apache.eventmesh.storage.redis.consumer.RedisConsumer;
+import org.apache.eventmesh.storage.redis.consumer.RedisConsumerTest;
+import org.apache.eventmesh.storage.redis.producer.RedisProducer;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class UnitTest extends AbstractRedisServer {
+
+ private RedisProducer redisProducer;
+
+ private RedisConsumer redisConsumer;
+
+ @Before
+ public void setup() {
+ redisProducer = new RedisProducer();
+ redisProducer.init(new Properties());
+ redisProducer.start();
+
+ redisConsumer = new RedisConsumer();
+ redisConsumer.init(new Properties());
+ redisConsumer.start();
+ }
+
+ @After
+ public void shutdown() {
+ redisProducer.shutdown();
+ redisConsumer.shutdown();
+ }
+
+ @Test
+ public void testPubSub() throws Exception {
+
+ final int expectedCount = 3;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+
+ redisConsumer.registerEventListener((cloudEvent, context) -> {
+ downLatch.countDown();
+ context.commit(EventMeshAction.CommitMessage);
+ }
+ );
+
+ final String topic = RedisConsumerTest.class.getSimpleName();
+
+ redisConsumer.subscribe(topic);
+
+ 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();
+
+ redisProducer.publish(cloudEvent, new SendCallback() {
+
+ @Override
+ public void onSuccess(SendResult sendResult) {
+
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+
+ }
+ });
+ }
+
+ Assert.assertTrue(downLatch.await(5, TimeUnit.MINUTES));
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumerTest.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumerTest.java
new file mode 100644
index 000000000..b29e034c0
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/consumer/RedisConsumerTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.redis.consumer;
+
+import org.apache.eventmesh.api.EventMeshAction;
+import org.apache.eventmesh.storage.redis.AbstractRedisServer;
+import org.apache.eventmesh.storage.redis.client.RedissonClient;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.redisson.api.RTopic;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class RedisConsumerTest extends AbstractRedisServer {
+
+ private RedisConsumer redisConsumer;
+
+ @Before
+ public void setup() {
+ redisConsumer = new RedisConsumer();
+ redisConsumer.init(new Properties());
+ redisConsumer.start();
+ }
+
+ @After
+ public void shutdown() {
+ redisConsumer.shutdown();
+ }
+
+ @Test
+ public void testSubscribe() throws Exception {
+
+ final int expectedCount = 3;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+
+ redisConsumer.registerEventListener((cloudEvent, context) -> {
+ downLatch.countDown();
+ context.commit(EventMeshAction.CommitMessage);
+ }
+ );
+
+ final String topic = RedisConsumerTest.class.getSimpleName();
+
+ redisConsumer.subscribe(topic);
+
+ RTopic redissonTopic = RedissonClient.INSTANCE.getTopic(topic);
+ 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();
+
+ redissonTopic.publish(cloudEvent);
+ }
+
+ Assert.assertTrue(downLatch.await(5, TimeUnit.MINUTES));
+
+ redisConsumer.unsubscribe(topic);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/producer/RedisProducerTest.java
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/producer/RedisProducerTest.java
new file mode 100644
index 000000000..946dcc830
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/java/org/apache/eventmesh/storage/redis/producer/RedisProducerTest.java
@@ -0,0 +1,110 @@
+/*
+ * 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.redis.producer;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.exception.OnExceptionContext;
+import org.apache.eventmesh.storage.redis.AbstractRedisServer;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.Properties;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+public class RedisProducerTest extends AbstractRedisServer {
+
+ private RedisProducer redisProducer;
+
+ @Before
+ public void setup() {
+ redisProducer = new RedisProducer();
+ redisProducer.init(new Properties());
+ redisProducer.start();
+ }
+
+ @After
+ public void shutdown() {
+ redisProducer.shutdown();
+ }
+
+ @Test
+ public void testPublish() throws Exception {
+ final int expectedCount = 3;
+ final CountDownLatch downLatch = new CountDownLatch(expectedCount);
+
+ for (int i = 0; i < expectedCount; i++) {
+ CloudEvent cloudEvent = CloudEventBuilder.v1()
+ .withId(String.valueOf(i))
+ .withTime(OffsetDateTime.now())
+ .withSource(URI.create("testsource"))
+ .withSubject(RedisProducerTest.class.getSimpleName())
+ .withType(String.class.getCanonicalName())
+ .withDataContentType("text/plain")
+ .withData("data".getBytes(StandardCharsets.UTF_8))
+ .build();
+
+ redisProducer.publish(cloudEvent, new SendCallback() {
+
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ downLatch.countDown();
+ Assert.assertEquals(cloudEvent.getId(),
sendResult.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
sendResult.getTopic());
+ }
+
+ @Override
+ public void onException(OnExceptionContext context) {
+ downLatch.countDown();
+ Assert.assertEquals(cloudEvent.getId(),
context.getMessageId());
+ Assert.assertEquals(cloudEvent.getSubject(),
context.getTopic());
+ }
+ });
+ }
+
+ downLatch.await();
+ }
+
+ @Test
+ public void testSendOneway() {
+
+ final String topic = RedisProducerTest.class.getSimpleName();
+
+ CloudEvent cloudEvent = CloudEventBuilder.v1()
+ .withId(UUID.randomUUID().toString())
+ .withTime(OffsetDateTime.now())
+ .withSource(URI.create("testsource"))
+ .withSubject(topic)
+ .withType(String.class.getCanonicalName())
+ .withDataContentType("text/plain")
+ .withData("data".getBytes())
+ .build();
+
+ redisProducer.sendOneway(cloudEvent);
+ }
+}
diff --git
a/eventmesh-storage/eventmesh-storage-redis/src/test/resources/redis-client.properties
b/eventmesh-storage/eventmesh-storage-redis/src/test/resources/redis-client.properties
new file mode 100644
index 000000000..7c89643f7
--- /dev/null
+++
b/eventmesh-storage/eventmesh-storage-redis/src/test/resources/redis-client.properties
@@ -0,0 +1,21 @@
+#
+# 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.
+#
+eventMesh.server.redis.serverAddress=redis://127.0.0.1:6379
+eventMesh.server.redis.serverType=SINGLE
+eventMesh.server.redis.serverMasterName=serverMasterName-success!!!
+eventMesh.server.redis.redisson.threads=2
+eventMesh.server.redis.redisson.nettyThreads=2
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
