This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 19dd2b1d5 [Feature #4032] Fix redis codec (#4301)
19dd2b1d5 is described below
commit 19dd2b1d5cc7abd327b808a3707ca4aaafce8b31
Author: Fabian Bao <[email protected]>
AuthorDate: Thu Aug 3 09:20:32 2023 +0800
[Feature #4032] Fix redis codec (#4301)
* fix: add cloudEventCodec for redis connector.
* fix: add cloudEventCodec for redis connector.
* fix import order.
* fix: add doc.
* fix: adjust cloudevent encode and decode.
---
.../eventmesh-connector-redis/build.gradle | 2 ++
.../redis/cloudevent/CloudEventCodec.java | 6 ++--
.../redis/sink/connector/RedisSinkConnector.java | 39 ++++++++++++++++++++--
.../source/connector/RedisSourceConnector.java | 29 ++++++++++++----
.../src/main/resources/sink-config.yml | 2 +-
.../src/main/resources/source-config.yml | 2 +-
.../storage/redis/cloudevent/CloudEventCodec.java | 1 -
7 files changed, 68 insertions(+), 13 deletions(-)
diff --git a/eventmesh-connectors/eventmesh-connector-redis/build.gradle
b/eventmesh-connectors/eventmesh-connector-redis/build.gradle
index 8b92940aa..42d3d4ced 100644
--- a/eventmesh-connectors/eventmesh-connector-redis/build.gradle
+++ b/eventmesh-connectors/eventmesh-connector-redis/build.gradle
@@ -20,6 +20,8 @@ dependencies {
implementation 'org.redisson:redisson:3.17.3'
+ api 'io.cloudevents:cloudevents-json-jackson'
+
compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
}
\ No newline at end of file
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/cloudevent/CloudEventCodec.java
similarity index 93%
copy from
eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
copy to
eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/cloudevent/CloudEventCodec.java
index 9db3c43dc..8fb347354 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/cloudevent/CloudEventCodec.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.eventmesh.storage.redis.cloudevent;
+package org.apache.eventmesh.connector.redis.cloudevent;
import org.redisson.client.codec.BaseCodec;
import org.redisson.client.protocol.Decoder;
@@ -26,7 +26,9 @@ import io.cloudevents.jackson.JsonFormat;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
-
+/**
+ * TODO: Duplicate with
org.apache.eventmesh.storage.redis.cloudevent.CloudEventCodec
+ */
public class CloudEventCodec extends BaseCodec {
private static final CloudEventCodec INSTANCE = new CloudEventCodec();
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
index 58466cc97..1193c0477 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/sink/connector/RedisSinkConnector.java
@@ -17,19 +17,24 @@
package org.apache.eventmesh.connector.redis.sink.connector;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
import org.apache.eventmesh.connector.redis.sink.config.RedisSinkConfig;
import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.data.ConnectRecord;
import org.apache.eventmesh.openconnect.api.sink.Sink;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.List;
import org.redisson.Redisson;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
-import lombok.extern.slf4j.Slf4j;
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+import lombok.extern.slf4j.Slf4j;
@Slf4j
public class RedisSinkConnector implements Sink {
@@ -50,6 +55,7 @@ public class RedisSinkConnector implements Sink {
this.sinkConfig = (RedisSinkConfig) config;
org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
redisConfig.useSingleServer().setAddress(sinkConfig.connectorConfig.getServer());
+ redisConfig.setCodec(CloudEventCodec.getInstance());
this.redissonClient = Redisson.create(redisConfig);
}
@@ -76,11 +82,40 @@ public class RedisSinkConnector implements Sink {
@Override
public void put(List<ConnectRecord> sinkRecords) {
for (ConnectRecord connectRecord : sinkRecords) {
+ CloudEvent event = convertRecordToEvent(connectRecord);
try {
- long publish = topic.publish(connectRecord);
+ long publish = topic.publish(event);
} catch (Exception e) {
log.error("[RedisSinkConnector] sendResult has error : ", e);
}
}
}
+
+ public CloudEvent convertRecordToEvent(ConnectRecord connectRecord) {
+ CloudEventBuilder cloudEventBuilder = CloudEventBuilder.v1()
+ .withData((byte[]) connectRecord.getData());
+ connectRecord.getExtensions().keySet().forEach(s -> {
+ switch (s) {
+ case "id":
+ cloudEventBuilder.withId(connectRecord.getExtension(s));
+ break;
+ case "topic":
+
cloudEventBuilder.withSubject(connectRecord.getExtension(s));
+ break;
+ case "source":
+ try {
+ cloudEventBuilder.withSource(new
URI(connectRecord.getExtension(s)));
+ } catch (URISyntaxException e) {
+ throw new RuntimeException(e);
+ }
+ break;
+ case "type":
+ cloudEventBuilder.withType(connectRecord.getExtension(s));
+ break;
+ default:
+ cloudEventBuilder.withExtension(s,
connectRecord.getExtension(s));
+ }
+ });
+ return cloudEventBuilder.build();
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
index eca1c23c3..ae0824e16 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/java/org/apache/eventmesh/connector/redis/source/connector/RedisSourceConnector.java
@@ -17,6 +17,7 @@
package org.apache.eventmesh.connector.redis.source.connector;
+import org.apache.eventmesh.connector.redis.cloudevent.CloudEventCodec;
import org.apache.eventmesh.connector.redis.source.config.RedisSourceConfig;
import org.apache.eventmesh.openconnect.api.config.Config;
import org.apache.eventmesh.openconnect.api.data.ConnectRecord;
@@ -24,6 +25,7 @@ import org.apache.eventmesh.openconnect.api.source.Source;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -32,7 +34,7 @@ import org.redisson.Redisson;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
-
+import io.cloudevents.CloudEvent;
public class RedisSourceConnector implements Source {
@@ -44,7 +46,7 @@ public class RedisSourceConnector implements Source {
private RedissonClient redissonClient;
- private BlockingQueue<ConnectRecord> queue;
+ private BlockingQueue<CloudEvent> queue;
@Override
public Class<? extends Config> configClass() {
@@ -56,6 +58,7 @@ public class RedisSourceConnector implements Source {
this.sourceConfig = (RedisSourceConfig) config;
org.redisson.config.Config redisConfig = new
org.redisson.config.Config();
redisConfig.useSingleServer().setAddress(sourceConfig.connectorConfig.getServer());
+ redisConfig.setCodec(CloudEventCodec.getInstance());
this.redissonClient = Redisson.create(redisConfig);
this.queue = new LinkedBlockingQueue<>(1000);
}
@@ -63,7 +66,7 @@ public class RedisSourceConnector implements Source {
@Override
public void start() throws Exception {
this.topic =
redissonClient.getTopic(sourceConfig.connectorConfig.getTopic());
- this.topic.addListener(ConnectRecord.class, (channel, msg) -> {
+ this.topic.addListener(CloudEvent.class, (channel, msg) -> {
queue.add(msg);
});
}
@@ -89,15 +92,29 @@ public class RedisSourceConnector implements Source {
List<ConnectRecord> connectRecords = new
ArrayList<>(DEFAULT_BATCH_SIZE);
for (int count = 0; count < DEFAULT_BATCH_SIZE; ++count) {
try {
- ConnectRecord connectRecord = queue.poll(3, TimeUnit.SECONDS);
- if (connectRecord == null) {
+ CloudEvent event = queue.poll(3, TimeUnit.SECONDS);
+ if (event == null) {
break;
}
- connectRecords.add(connectRecord);
+
+ connectRecords.add(convertEventToRecord(event));
} catch (InterruptedException e) {
break;
}
}
return connectRecords;
}
+
+ public ConnectRecord convertEventToRecord(CloudEvent event) {
+ byte[] body = Objects.requireNonNull(event.getData()).toBytes();
+ ConnectRecord connectRecord = new ConnectRecord(null, null,
System.currentTimeMillis(), body);
+ for (String extensionName : event.getExtensionNames()) {
+ connectRecord.addExtension(extensionName,
Objects.requireNonNull(event.getExtension(extensionName)).toString());
+ }
+ connectRecord.addExtension("id", event.getId());
+ connectRecord.addExtension("topic", event.getSubject());
+ connectRecord.addExtension("source", event.getSource().toString());
+ connectRecord.addExtension("type", event.getType());
+ return connectRecord;
+ }
}
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/sink-config.yml
b/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/sink-config.yml
index 920f92e22..596006a2a 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/sink-config.yml
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/sink-config.yml
@@ -27,5 +27,5 @@ pubSubConfig:
connectorConfig:
connectorName: redisSink
server: redis://127.0.0.1:6379
- topic: TopicTest
+ topic: SinkTopic
diff --git
a/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/source-config.yml
b/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/source-config.yml
index 1f08068db..5d31ad0fc 100644
---
a/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/source-config.yml
+++
b/eventmesh-connectors/eventmesh-connector-redis/src/main/resources/source-config.yml
@@ -27,4 +27,4 @@ pubSubConfig:
connectorConfig:
connectorName: redisSource
server: redis://127.0.0.1:6379
- topic: TopicTest
+ topic: SourceTopic
diff --git
a/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
b/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
index 9db3c43dc..e05903d46 100644
---
a/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
+++
b/eventmesh-storage-plugin/eventmesh-storage-redis/src/main/java/org/apache/eventmesh/storage/redis/cloudevent/CloudEventCodec.java
@@ -26,7 +26,6 @@ import io.cloudevents.jackson.JsonFormat;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
-
public class CloudEventCodec extends BaseCodec {
private static final CloudEventCodec INSTANCE = new CloudEventCodec();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
