added support for lettuce reactive api

Author: jruaux

src/main/java/com/redislabs/riot/cli/RedisDriver.java

@@ -1,5 +1,5 @@
 package com.redislabs.riot.cli;
 
 public enum RedisDriver {
-	Jedis, Lettuce
+	Jedis, LettuceAsync, LettuceReactive
 }

src/main/java/com/redislabs/riot/cli/RootCommand.java

@@ -31,7 +31,6 @@ public class RootCommand extends AbstractCommand {
 	 */
 	@Option(names = "--spring.output.ansi.enabled", hidden = true)
 	private String ansiEnabled;
-	@Getter
 	@Option(names = "--host", description = "Redis server host. (default: localhost).")
 	private InetAddress host;
 	@Option(names = "--port", description = "Redis server port. (default: ${DEFAULT-VALUE}).")
@@ -56,6 +55,9 @@ public class RootCommand extends AbstractCommand {
 	private int database = 0;
 	@Option(names = "--client-name", description = "Redis client name.")
 	private String clientName;
+	@Getter
+	@Option(names = "--driver", description = "Redis driver: ${COMPLETION-CANDIDATES}. (default: ${DEFAULT-VALUE})")
+	private RedisDriver driver = RedisDriver.Jedis;
 
 	public String getHostname() {
 		if (host != null) {

src/main/java/com/redislabs/riot/cli/in/AbstractRedisImportWriterCommand.java

@@ -4,27 +4,26 @@
 
 import org.springframework.batch.item.ItemWriter;
 
-import com.redislabs.riot.cli.RedisDriver;
+import com.redislabs.lettusearch.RediSearchClient;
 import com.redislabs.riot.redis.writer.AbstractRedisDataStructureItemWriter;
 import com.redislabs.riot.redis.writer.JedisWriter;
-import com.redislabs.riot.redis.writer.LettuceWriter;
-
-import picocli.CommandLine.Option;
+import com.redislabs.riot.redis.writer.LettuceAsyncWriter;
+import com.redislabs.riot.redis.writer.LettuceReactiveWriter;
 
 public abstract class AbstractRedisImportWriterCommand extends AbstractImportWriterCommand {
 
-	@Option(names = "--driver", description = "Redis driver: ${COMPLETION-CANDIDATES}. (default: ${DEFAULT-VALUE})")
-	private RedisDriver driver = RedisDriver.Jedis;
-
 	@Override
 	protected ItemWriter<Map<String, Object>> writer() {
 		AbstractRedisDataStructureItemWriter itemWriter = redisItemWriter();
 		itemWriter.setConverter(redisConverter());
-		switch (driver) {
-		case Jedis:
-			return new JedisWriter(getRoot().jedisPool(), itemWriter);
+		switch (getRoot().getDriver()) {
+		case LettuceAsync:
+			return new LettuceAsyncWriter(getRoot().lettucePool(), itemWriter);
+		case LettuceReactive:
+			RediSearchClient client = getRoot().lettuceClient();
+			return new LettuceReactiveWriter(client.connect(), itemWriter);
 		default:
-			return new LettuceWriter(getRoot().lettucePool(), itemWriter);
+			return new JedisWriter(getRoot().jedisPool(), itemWriter);
 		}
 	}
 

src/main/java/com/redislabs/riot/cli/in/redis/AbstractRediSearchImport.java

@@ -5,7 +5,7 @@
 import org.springframework.batch.item.ItemWriter;
 
 import com.redislabs.riot.cli.in.AbstractImportWriterCommand;
-import com.redislabs.riot.redis.writer.LettuceWriter;
+import com.redislabs.riot.redis.writer.LettuceAsyncWriter;
 import com.redislabs.riot.redis.writer.search.AbstractLettuSearchItemWriter;
 
 import lombok.Getter;
@@ -30,7 +30,7 @@ protected String getTargetDescription() {
 
 	@Override
 	protected ItemWriter<Map<String, Object>> writer() {
-		return new LettuceWriter(getRoot().lettucePool(), lettuceItemWriter());
+		return new LettuceAsyncWriter(getRoot().lettucePool(), lettuceItemWriter());
 	}
 
 	protected abstract ItemWriter<Map<String, Object>> jedisSearchWriter();

src/main/java/com/redislabs/riot/cli/in/redis/SearchImport.java

@@ -6,7 +6,9 @@
 
 import com.redislabs.lettusearch.search.AddOptions;
 import com.redislabs.lettusearch.search.Language;
+import com.redislabs.riot.redis.writer.search.AbstractLettuSearchItemWriter;
 import com.redislabs.riot.redis.writer.search.JedisSearchWriter;
+import com.redislabs.riot.redis.writer.search.LettuSearchAddPayloadWriter;
 import com.redislabs.riot.redis.writer.search.LettuSearchAddWriter;
 
 import io.redisearch.client.AddOptions.ReplacementPolicy;
@@ -29,20 +31,27 @@ public class SearchImport extends AbstractRediSearchImport {
 	@Option(names = "--if", description = "Applicable only in conjunction with REPLACE and optionally PARTIAL. Update the document only if a boolean expression applies to the document before the update.")
 	private String ifCondition;
 	@Option(names = "--score", description = "Name of the field to use for scores.")
-	private String score;
+	private String scoreField;
 	@Option(names = "--default-score", description = "Default score to use when score field is not present. (default: ${DEFAULT-VALUE}).")
 	private double defaultScore = 1d;
 	@Option(names = "--payload", description = "Name of the field containing the payload")
-	private String payload;
+	private String payloadField;
 
 	@Override
-	protected LettuSearchAddWriter rediSearchItemWriter() {
-		LettuSearchAddWriter writer = new LettuSearchAddWriter();
+	protected AbstractLettuSearchItemWriter rediSearchItemWriter() {
+		AddOptions options = AddOptions.builder().ifCondition(ifCondition).language(language).noSave(noSave)
+				.replace(replace).replacePartial(partial).build();
+		if (payloadField == null) {
+			LettuSearchAddWriter writer = new LettuSearchAddWriter();
+			writer.setDefaultScore(defaultScore);
+			writer.setOptions(options);
+			writer.setScoreField(scoreField);
+			return writer;
+		}
+		LettuSearchAddPayloadWriter writer = new LettuSearchAddPayloadWriter();
 		writer.setDefaultScore(defaultScore);
-		writer.setOptions(AddOptions.builder().ifCondition(ifCondition).language(language).noSave(noSave)
-				.replace(replace).replacePartial(partial).build());
-		writer.setPayloadField(payload);
-		writer.setScoreField(score);
+		writer.setOptions(options);
+		writer.setScoreField(scoreField);
 		return writer;
 	}
 
@@ -53,8 +62,8 @@ protected ItemWriter<Map<String, Object>> jedisSearchWriter() {
 		writer.setClient(new Client(getIndex(), pool));
 		writer.setConverter(redisConverter());
 		writer.setDefaultScore((float) defaultScore);
-		writer.setScoreField(score);
-		writer.setPayloadField(payload);
+		writer.setScoreField(scoreField);
+		writer.setPayloadField(payloadField);
 		io.redisearch.client.AddOptions options = new io.redisearch.client.AddOptions();
 		if (language != null) {
 			options.setLanguage(language.name());

src/main/java/com/redislabs/riot/cli/in/redis/StreamImport.java

@@ -1,6 +1,10 @@
 package com.redislabs.riot.cli.in.redis;
 
 import com.redislabs.riot.cli.in.AbstractRedisImportWriterCommand;
+import com.redislabs.riot.redis.writer.AbstractRedisDataStructureItemWriter;
+import com.redislabs.riot.redis.writer.StreamIdMaxlenWriter;
+import com.redislabs.riot.redis.writer.StreamIdWriter;
+import com.redislabs.riot.redis.writer.StreamMaxlenWriter;
 import com.redislabs.riot.redis.writer.StreamWriter;
 
 import picocli.CommandLine.Command;
@@ -18,7 +22,7 @@ public class StreamImport extends AbstractRedisImportWriterCommand {
 	@Option(names = "--maxlen", description = "Limit stream to maxlen entries.")
 	private Long maxlen;
 	@Option(names = "--id", description = "Field used for stream entry IDs.")
-	private String id;
+	private String idField;
 
 	@Override
 	protected String getKeyspace() {
@@ -31,10 +35,24 @@ protected String[] getKeys() {
 	}
 
 	@Override
-	protected StreamWriter redisItemWriter() {
-		StreamWriter writer = new StreamWriter();
+	protected AbstractRedisDataStructureItemWriter redisItemWriter() {
+		if (idField == null) {
+			if (maxlen == null) {
+				return new StreamWriter();
+			}
+			StreamMaxlenWriter writer = new StreamMaxlenWriter();
+			writer.setMaxlen(maxlen);
+			writer.setApproximateTrimming(approximateTrimming);
+			return writer;
+		}
+		if (maxlen == null) {
+			StreamIdWriter writer = new StreamIdWriter();
+			writer.setIdField(idField);
+			return writer;
+		}
+		StreamIdMaxlenWriter writer = new StreamIdMaxlenWriter();
 		writer.setApproximateTrimming(approximateTrimming);
-		writer.setIdField(id);
+		writer.setIdField(idField);
 		writer.setMaxlen(maxlen);
 		return writer;
 	}

src/main/java/com/redislabs/riot/cli/in/redis/SuggestImport.java

@@ -5,6 +5,7 @@
 import org.springframework.batch.item.ItemWriter;
 
 import com.redislabs.riot.redis.writer.search.AbstractLettuSearchItemWriter;
+import com.redislabs.riot.redis.writer.search.LettuSearchSuggestPayloadWriter;
 import com.redislabs.riot.redis.writer.search.LettuSearchSuggestWriter;
 
 import picocli.CommandLine.Command;
@@ -14,24 +15,32 @@
 public class SuggestImport extends AbstractRediSearchImport {
 
 	@Option(names = "--score", description = "Name of the field to use for scores.")
-	private String score;
+	private String scoreField;
 	@Option(names = "--default-score", description = "Default score to use when score field is not present. (default: ${DEFAULT-VALUE}).")
 	private Double defaultScore = 1d;
 	@Option(names = "--increment", description = "Use increment to set value")
 	private boolean increment;
 	@Option(names = "--suggest", description = "Name of the field containing the suggestion")
-	private String suggest;
+	private String suggestField;
 	@Option(names = "--payload", description = "Name of the field containing the payload")
-	private String payload;
+	private String payloadField;
 
 	@Override
 	protected AbstractLettuSearchItemWriter rediSearchItemWriter() {
-		LettuSearchSuggestWriter writer = new LettuSearchSuggestWriter();
+		if (payloadField == null) {
+			LettuSearchSuggestWriter writer = new LettuSearchSuggestWriter();
+			writer.setDefaultScore(defaultScore);
+			writer.setField(suggestField);
+			writer.setIncrement(increment);
+			writer.setScoreField(scoreField);
+			return writer;
+		}
+		LettuSearchSuggestPayloadWriter writer = new LettuSearchSuggestPayloadWriter();
 		writer.setDefaultScore(defaultScore);
-		writer.setField(suggest);
+		writer.setField(suggestField);
 		writer.setIncrement(increment);
-		writer.setPayloadField(payload);
-		writer.setScoreField(score);
+		writer.setPayloadField(payloadField);
+		writer.setScoreField(scoreField);
 		return writer;
 	}
 

src/main/java/com/redislabs/riot/redis/writer/AbstractCollectionRedisItemWriter.java

@@ -4,7 +4,9 @@
 
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import lombok.Setter;
+import reactor.core.publisher.Mono;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Response;
 
@@ -32,4 +34,12 @@ protected RedisFuture<?> write(RedisAsyncCommands<String, String> commands, Stri
 	protected abstract RedisFuture<?> write(RedisAsyncCommands<String, String> commands, String key, String member,
 			Map<String, Object> item);
 
+	@Override
+	protected Mono<?> write(RedisReactiveCommands<String, String> commands, String key, Map<String, Object> item) {
+		return write(commands, key, member(item), item);
+	}
+
+	protected abstract Mono<?> write(RedisReactiveCommands<String, String> commands, String key, String member,
+			Map<String, Object> item);
+
 }

src/main/java/com/redislabs/riot/redis/writer/AbstractRedisDataStructureItemWriter.java

@@ -4,6 +4,8 @@
 
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
+import reactor.core.publisher.Mono;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Response;
 
@@ -25,4 +27,12 @@ public RedisFuture<?> write(RedisAsyncCommands<String, String> commands, Map<Str
 	protected abstract RedisFuture<?> write(RedisAsyncCommands<String, String> commands, String key,
 			Map<String, Object> item);
 
+	@Override
+	public Mono<?> write(RedisReactiveCommands<String, String> commands, Map<String, Object> item) {
+		return write(commands, key(item), item);
+	}
+
+	protected abstract Mono<?> write(RedisReactiveCommands<String, String> commands, String key,
+			Map<String, Object> item);
+
 }

src/main/java/com/redislabs/riot/redis/writer/GeoWriter.java

@@ -4,46 +4,80 @@
 
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
 import lombok.Setter;
+import lombok.Value;
+import reactor.core.publisher.Mono;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Response;
 
 public class GeoWriter extends AbstractCollectionRedisItemWriter {
 
+	@Value
+	private static class Point {
+		private Double longitude;
+		private Double latitude;
+
+		public boolean isInvalid() {
+			return longitude == null || latitude == null;
+		}
+
+	}
+
 	@Setter
 	private String longitudeField;
 	@Setter
 	private String latitudeField;
 
 	@Override
 	protected Response<Long> write(Pipeline pipeline, String key, String member, Map<String, Object> item) {
-		Object longitude = item.get(longitudeField);
-		if (longitude == null || longitude.equals("")) {
+		Point point = point(item);
+		if (point.isInvalid()) {
 			return null;
 		}
-		Object latitude = item.get(latitudeField);
-		if (latitude == null || latitude.equals("")) {
-			return null;
+		return pipeline.geoadd(key, point.getLongitude(), point.getLatitude(), member);
+	}
+
+	private Point point(Map<String, Object> item) {
+		return new Point(longitude(item), latitude(item));
+	}
+
+	private Double coordinate(Map<String, Object> item, String field) {
+		if (item.containsKey(field)) {
+			Object value = item.get(field);
+			if (value != null && !"".equals(value)) {
+				return convert(value, Double.class);
+			}
 		}
-		double lon = convert(longitude, Double.class);
-		double lat = convert(latitude, Double.class);
-		return pipeline.geoadd(key, lon, lat, member);
+		return null;
+	}
+
+	private Double longitude(Map<String, Object> item) {
+		return coordinate(item, longitudeField);
+	}
+
+	private Double latitude(Map<String, Object> item) {
+		return coordinate(item, latitudeField);
 	}
 
 	@Override
 	protected RedisFuture<?> write(RedisAsyncCommands<String, String> commands, String key, String member,
 			Map<String, Object> item) {
-		Object longitude = item.get(longitudeField);
-		if (longitude == null || longitude.equals("")) {
+		Point point = point(item);
+		if (point.isInvalid()) {
 			return null;
 		}
-		Object latitude = item.get(latitudeField);
-		if (latitude == null || latitude.equals("")) {
+		return commands.geoadd(key, point.getLongitude(), point.getLatitude(), member);
+	}
+
+	@Override
+	protected Mono<?> write(RedisReactiveCommands<String, String> commands, String key, String member,
+			Map<String, Object> item) {
+		Point point = point(item);
+		if (point.isInvalid()) {
 			return null;
 		}
-		double lon = convert(longitude, Double.class);
-		double lat = convert(latitude, Double.class);
-		return commands.geoadd(key, lon, lat, member);
+		return commands.geoadd(key, point.getLongitude(), point.getLatitude(), member);
 	}
 
 }

src/main/java/com/redislabs/riot/redis/writer/HashWriter.java

@@ -4,6 +4,8 @@
 
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.api.async.RedisAsyncCommands;
+import io.lettuce.core.api.reactive.RedisReactiveCommands;
+import reactor.core.publisher.Mono;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Response;
 
@@ -19,4 +21,9 @@ protected RedisFuture<?> write(RedisAsyncCommands<String, String> commands, Stri
 		return commands.hmset(key, stringMap(item));
 	}
 
+	@Override
+	protected Mono<String> write(RedisReactiveCommands<String, String> commands, String key, Map<String, Object> item) {
+		return commands.hmset(key, stringMap(item));
+	}
+
 }