Fixed KeyValueProcessor expression parsing and added test case

connectors/riot-redis/src/test/java/com/redis/riot/redis/TestReplicate.java

@@ -79,6 +79,19 @@ void replicate(RedisServer container) throws Throwable {
         Assertions.assertEquals(sourceSize, targetSync.dbsize());
     }
 
+    @ParameterizedTest
+    @MethodSource("containers")
+    void replicateKeyProcessor(RedisServer container) throws Throwable {
+        dataGenerator(container).build().call();
+        RedisServerCommands<String, String> sync = sync(container);
+        Long sourceSize = sync.dbsize();
+        Assertions.assertTrue(sourceSize > 0);
+        execute("replicate-key-processor", container, r -> configureReplicateCommand(r,false));
+        Assertions.assertEquals(sourceSize, targetSync.dbsize());
+        RedisStringCommands<String,String> stringCommands = sync(container);
+        Assertions.assertEquals(stringCommands.get("string:123"), targetSync.get("0:string:123"));
+    }
+
     @ParameterizedTest
     @MethodSource("containers")
     public void replicateLive(RedisServer container) throws Exception {

connectors/riot-redis/src/test/resources/replicate-key-processor

@@ -0,0 +1 @@
+riot-redis -h source -p 6379 replicate -h target -p 6380 --batch 10 --key-process="#{#src.database}:#{key}"

core/riot-core/src/main/java/com/redislabs/riot/KeyValueProcessorOptions.java

@@ -6,19 +6,23 @@
 import org.springframework.batch.item.redis.support.KeyValue;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
+import org.springframework.expression.common.TemplateParserContext;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 import picocli.CommandLine.Option;
 
 @Data
 public class KeyValueProcessorOptions {
 
     @Option(names = "--key-process", description = "SpEL expression to transform each key", paramLabel = "<exp>")
-    private Expression expression;
+    private String keyProcessor;
 
     public <T extends KeyValue<?>> ItemProcessor<T, T> processor(RedisOptions sourceRedis, RedisOptions targetRedis) {
-        if (expression == null) {
+        if (keyProcessor == null) {
             return null;
         }
+        SpelExpressionParser parser = new SpelExpressionParser();
+        Expression expression = parser.parseExpression(keyProcessor, new TemplateParserContext());
         EvaluationContext context = new StandardEvaluationContext();
         context.setVariable("src", sourceRedis.uris().get(0));
         context.setVariable("dest", targetRedis.uris().get(0));

core/riot-core/src/test/java/com/redislabs/riot/ProcessorsTest.java

@@ -6,7 +6,6 @@
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.redis.support.KeyValue;
 import org.springframework.expression.Expression;
-import org.springframework.expression.common.TemplateParserContext;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.expression.spel.support.StandardEvaluationContext;
 
@@ -25,9 +24,7 @@ public void testKeyValueProcessor() {
     @Test
     public void testKeyValueProcessorOptions() throws Exception {
         KeyValueProcessorOptions options = new KeyValueProcessorOptions();
-        SpelExpressionParser parser = new SpelExpressionParser();
-        Expression expression = parser.parseExpression("#{#src.database}:#{key}:#{#src.host}", new TemplateParserContext());
-        options.setExpression(expression);
+        options.setKeyProcessor("#{#src.database}:#{key}:#{#src.host}");
         RedisOptions src = new RedisOptions();
         src.setDatabase(14);
         src.setHost("srchost");