[FLINK-15571] Add basic Redis Streams sink.

Author: eskabetxe

flink-connector-redis-streams/pom.xml

@@ -0,0 +1,36 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>flink-connector-redis-parent</artifactId>
+        <groupId>org.apache.flink</groupId>
+        <version>3.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>flink-connector-redis-streams</artifactId>
+    <name>Flink : Connectors : Redis : Streams</name>
+
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>redis.clients</groupId>
+            <artifactId>jedis</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-base</artifactId>
+            <version>${flink.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-streaming-java</artifactId>
+        </dependency>
+    </dependencies>
+
+
+</project>

flink-connector-redis-streams/src/main/java/org/apache/flink/connector/redis/streams/sink/RedisSerializer.java

@@ -0,0 +1,32 @@
+/*
+ * 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.flink.connector.redis.streams.sink;
+
+import org.apache.flink.api.common.functions.Function;
+import org.apache.flink.connector.redis.streams.sink.command.RedisCommand;
+
+import java.io.Serializable;
+
+/**
+ * Function that creates the description how the input data should be mapped to redis type.
+ *
+ * @param <T> The type of the element handled by this {@code RedisSerializer}
+ */
+public interface RedisSerializer<T> extends Function, Serializable {
+
+    RedisCommand getMessage(T input);
+}

flink-connector-redis-streams/src/main/java/org/apache/flink/connector/redis/streams/sink/RedisStreamsSink.java

@@ -0,0 +1,43 @@
+/*
+ * 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.flink.connector.redis.streams.sink;
+
+import org.apache.flink.api.connector.sink2.Sink;
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.redis.streams.sink.config.JedisConfig;
+import org.apache.flink.connector.redis.streams.sink.connection.JedisConnector;
+import org.apache.flink.connector.redis.streams.sink.connection.JedisConnectorBuilder;
+
+import java.io.IOException;
+
+public class RedisStreamsSink<T> implements Sink<T> {
+
+    private final JedisConfig jedisConfig;
+    private final RedisSerializer<T> serializer;
+
+    public RedisStreamsSink(JedisConfig jedisConfig, RedisSerializer<T> serializer) {
+        this.jedisConfig = jedisConfig;
+        this.serializer = serializer;
+    }
+
+    @Override
+    public SinkWriter<T> createWriter(InitContext initContext) throws IOException {
+        JedisConnector connection = JedisConnectorBuilder.build(jedisConfig);
+        return new RedisStreamsWriter<>(connection, this.serializer);
+    }
+
+}

flink-connector-redis-streams/src/main/java/org/apache/flink/connector/redis/streams/sink/RedisStreamsWriter.java

@@ -0,0 +1,61 @@
+/*
+ * 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.flink.connector.redis.streams.sink;
+
+import org.apache.flink.api.connector.sink2.SinkWriter;
+import org.apache.flink.connector.redis.streams.sink.command.RedisCommand;
+import org.apache.flink.connector.redis.streams.sink.connection.JedisConnector;
+
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Queue;
+
+public class RedisStreamsWriter<T> implements SinkWriter<T> {
+
+    private final JedisConnector jedisConnector;
+    private final RedisSerializer<T> serializer;
+    private final Queue<RedisCommand> queue = new ArrayDeque<>();
+
+    public RedisStreamsWriter(JedisConnector jedisConnector, RedisSerializer<T> serializer) {
+        this.jedisConnector = jedisConnector;
+        this.serializer = serializer;
+    }
+
+
+    @Override
+    public void write(T input, Context context) throws IOException, InterruptedException {
+        RedisCommand message = serializer.getMessage(input);
+        queue.add(message);
+    }
+
+    @Override
+    public void flush(boolean endOfInput) throws IOException, InterruptedException {
+        flush();
+    }
+
+    private void flush() {
+        while(!this.queue.isEmpty()) {
+            RedisCommand element = this.queue.remove();
+            element.send(this.jedisConnector);
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        jedisConnector.close();
+    }
+}

flink-connector-redis-streams/src/main/java/org/apache/flink/connector/redis/streams/sink/command/RedisCommand.java

@@ -0,0 +1,26 @@
+/*
+ * 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.flink.connector.redis.streams.sink.command;
+
+import org.apache.flink.connector.redis.streams.sink.connection.JedisConnector;
+
+import java.io.Serializable;
+
+public interface RedisCommand extends Serializable {
+    void send(JedisConnector connector);
+
+}

flink-connector-redis-streams/src/main/java/org/apache/flink/connector/redis/streams/sink/command/StreamRedisCommand.java

@@ -0,0 +1,61 @@
+/*
+ * 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.flink.connector.redis.streams.sink.command;
+
+
+import org.apache.flink.connector.redis.streams.sink.connection.JedisConnector;
+import redis.clients.jedis.StreamEntryID;
+
+import java.util.Map;
+
+public class StreamRedisCommand implements RedisCommand {
+    public final String key;
+    public final Map<String, String> value;
+
+    private StreamRedisCommand(String key, Map<String, String> value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public void send(JedisConnector connector) {
+        connector.getJedisCommands().xadd(key, StreamEntryID.NEW_ENTRY, value);
+    }
+
+    public static class Builder {
+        private String key;
+        private Map<String, String> value;
+
+        public Builder withKey(String key) {
+            this.key = key;
+            return this;
+        }
+
+        public Builder withValue(Map<String, String> value) {
+            this.value = value;
+            return this;
+        }
+
+        public StreamRedisCommand build() {
+            return new StreamRedisCommand(key, value);
+        }
+    }
+}