test: Added read-threads tests

plugins/riot/src/test/java/com/redis/riot/ProcessorTests.java

@@ -14,9 +14,9 @@
 
 import com.redis.riot.core.processor.StringToMapFunction;
 import com.redis.riot.function.KeyValueMap;
-import com.redis.spring.batch.Range;
 import com.redis.spring.batch.item.redis.common.DataType;
 import com.redis.spring.batch.item.redis.common.KeyValue;
+import com.redis.spring.batch.item.redis.common.Range;
 
 import io.lettuce.core.cluster.SlotHash;
 import io.lettuce.core.codec.StringCodec;
@@ -40,7 +40,7 @@ void keyFilter() {
 	@Test
 	void slotExact() {
 		KeyFilterArgs options = new KeyFilterArgs();
-		options.setSlots(Arrays.asList(Range.of(7638)));
+		options.setSlots(Arrays.asList(new Range(7638, 7638)));
 		Predicate<String> predicate = keyFilter(options);
 		assertTrue(predicate.test("abc"));
 		assertFalse(predicate.test("abcd"));
@@ -53,21 +53,21 @@ void slotRange() {
 		Predicate<String> unbounded = keyFilter(options);
 		assertTrue(unbounded.test("foo"));
 		assertTrue(unbounded.test("foo1"));
-		options.setSlots(slotRangeList(999999, 99999));
+		options.setSlots(slotRangeList(999999, 999999));
 		Predicate<String> is999999 = keyFilter(options);
 		assertFalse(is999999.test("foo"));
 	}
 
 	private List<Range> slotRangeList(int start, int end) {
-		return Arrays.asList(Range.of(start, end));
+		return Arrays.asList(new Range(start, end));
 	}
 
 	@Test
 	void kitchenSink() {
 		KeyFilterArgs options = new KeyFilterArgs();
 		options.setExcludes(Arrays.asList("foo"));
 		options.setIncludes(Arrays.asList("foo1"));
-		options.setSlots(Arrays.asList(Range.of(0, SlotHash.SLOT_COUNT)));
+		options.setSlots(Arrays.asList(new Range(0, SlotHash.SLOT_COUNT)));
 		Predicate<String> predicate = keyFilter(options);
 		assertFalse(predicate.test("foo"));
 		assertFalse(predicate.test("bar"));

plugins/riot/src/test/java/com/redis/riot/RedisArgsTests.java

@@ -3,12 +3,14 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import com.redis.spring.batch.item.redis.common.Range;
+
 import io.lettuce.core.RedisURI;
 
 class RedisArgsTests {
 
 	@Test
-	void testSimpleRedisArgsURI() {
+	void simpleRedisArgsURI() {
 		SimpleRedisArgs args = new SimpleRedisArgs();
 		RedisURI baseUri = RedisURI.create("redis://localhost");
 		args.setUri(baseUri);
@@ -20,12 +22,20 @@ void testSimpleRedisArgsURI() {
 	}
 
 	@Test
-	void testRedisURIParser() {
+	void parseRedisURI() {
 		String host = "blah";
 		int port = 123;
-		RedisURI uri = Riot.parseRedisURI(host + ":" + port);
+		RedisURI uri = new RedisURIConverter().convert(host + ":" + port);
 		Assertions.assertEquals(host, uri.getHost());
 		Assertions.assertEquals(port, uri.getPort());
 	}
 
+	@Test
+	void parseRange() {
+		RangeConverter converter = new RangeConverter();
+		Assertions.assertEquals(new Range(123, 123), converter.convert("123"));
+		Assertions.assertEquals(new Range(0, 123), converter.convert("0-123"));
+		Assertions.assertEquals(new Range(123), converter.convert("123-"));
+	}
+
 }

plugins/riot/src/test/java/com/redis/riot/RiotTests.java

@@ -18,9 +18,9 @@
 import com.redis.lettucemod.api.StatefulRedisModulesConnection;
 import com.redis.riot.Replicate.CompareMode;
 import com.redis.riot.core.ProgressStyle;
-import com.redis.spring.batch.Range;
 import com.redis.spring.batch.item.redis.RedisItemReader.ReaderMode;
 import com.redis.spring.batch.item.redis.common.DataType;
+import com.redis.spring.batch.item.redis.common.Range;
 import com.redis.spring.batch.item.redis.gen.GeneratorItemReader;
 
 import io.lettuce.core.cluster.SlotHash;
@@ -116,7 +116,7 @@ void replicateBinaryKeyLive(TestInfo info) throws Exception {
 				.connection(targetRedisClient, ByteArrayCodec.INSTANCE);
 		enableKeyspaceNotifications();
 		Executors.newSingleThreadExecutor().execute(() -> {
-			awaitUntilSubscribers();
+			awaitSubscribers();
 			connection.sync().set(key, value);
 		});
 		Replicate replication = new Replicate();
@@ -132,10 +132,10 @@ void filterKeySlot(TestInfo info) throws Exception {
 		Replicate replication = new Replicate();
 		replication.getSourceRedisReaderArgs().setMode(ReaderMode.LIVE);
 		replication.setCompareMode(CompareMode.NONE);
-		replication.getSourceRedisReaderArgs().getKeyFilterArgs().setSlots(Arrays.asList(Range.of(0, 8000)));
+		replication.getSourceRedisReaderArgs().getKeyFilterArgs().setSlots(Arrays.asList(new Range(0, 8000)));
 		generateAsync(info, generator(100));
 		execute(replication, info);
-		awaitUntilNoSubscribers();
+		awaitNoSubscribers();
 		Assertions.assertTrue(targetRedisCommands.keys("*").stream().map(SlotHash::getSlot).allMatch(between(0, 8000)));
 	}
 

plugins/riot/src/test/java/com/redis/riot/StackRiotTests.java

@@ -627,8 +627,8 @@ void replicate(TestInfo info) throws Throwable {
 	}
 
 	@Test
-	void replicateNoStreamIds(TestInfo info) throws Throwable {
-		String filename = "replicate-no-stream-ids";
+	void replicateNoStreamId(TestInfo info) throws Throwable {
+		String filename = "replicate-no-stream-id";
 		generate(info, generator(73));
 		Assertions.assertTrue(redisCommands.dbsize() > 0);
 		execute(info, filename);
@@ -644,8 +644,8 @@ void replicateNoStreamIds(TestInfo info) throws Throwable {
 	}
 
 	@Test
-	void replicateNoStreamIdsPrune(TestInfo info) throws Throwable {
-		String filename = "replicate-no-stream-ids-prune";
+	void replicateNoStreamIdPrune(TestInfo info) throws Throwable {
+		String filename = "replicate-no-stream-id-prune";
 		generate(info, generator(73));
 		String emptyStream = "stream:empty";
 		redisCommands.xadd(emptyStream, Map.of("field", "value"));
@@ -709,6 +709,11 @@ void replicateLiveThreads(TestInfo info) throws Exception {
 	void replicateLive(TestInfo info) throws Exception {
 		runLiveReplication(info, "replicate-live");
 	}
+
+	@Test
+	void replicateLiveReadThreads(TestInfo info) throws Exception {
+		runLiveReplication(info, "replicate-live-read-threads");
+	}
 
 	@Test
 	void replicateLiveKeySlot(TestInfo info) throws Exception {

plugins/riot/src/test/resources/faker-sadd

@@ -1 +1 @@
-riot faker name="GameOfThrones.character" --count 1000 sadd --keyspace got:characters --members name
+riot faker name="GameOfThrones.character" --count 1000 sadd --keyspace got:characters --member name

plugins/riot/src/test/resources/faker-tsadd-options

@@ -1 +1 @@
-riot faker value="Random.next_double" future="BackToTheFuture.character" lebowski="Lebowski.character" --count 1000 --sleep 1 ts.add --keyspace ts:gen --key future lebowski --value value --labels character1=future character2=lebowski
+riot faker value="Random.next_double" future="BackToTheFuture.character" lebowski="Lebowski.character" --count 1000 --sleep 1 ts.add --keyspace ts:gen --key future lebowski --value value --label character1=future character2=lebowski

plugins/riot/src/test/resources/faker-zadd

@@ -1 +1 @@
-riot faker ip="number.digits '4'" lease="number.digits '2'" time="number.digits '5'" zadd --keyspace leases --key ip --members lease --score=time
+riot faker ip="number.digits '4'" lease="number.digits '2'" time="number.digits '5'" zadd --keyspace leases --key ip --member lease --score=time

plugins/riot/src/test/resources/file-import-geoadd

@@ -1 +1 @@
-riot file-import http://storage.googleapis.com/jrx/airports.csv --header --skip-limit 3 geoadd --keyspace airportgeo --members AirportID --lon Longitude --lat Latitude
+riot file-import http://storage.googleapis.com/jrx/airports.csv --header --skip-limit 3 geoadd --keyspace airportgeo --member AirportID --lon Longitude --lat Latitude

plugins/riot/src/test/resources/file-import-hset-sadd

@@ -1 +1 @@
-riot file-import http://storage.googleapis.com/jrx/beers.csv --header hset --keyspace beer --key id sadd --keyspace breweries --members brewery_id
+riot file-import http://storage.googleapis.com/jrx/beers.csv --header hset --keyspace beer --key id sadd --keyspace breweries --member brewery_id

plugins/riot/src/test/resources/generate

@@ -1 +1 @@
-riot generate --types string hash json timeseries
+riot generate --type string hash json timeseries

plugins/riot/src/test/resources/generate-hash-index

@@ -1 +1 @@
-riot generate --types hash --index hashIdx
+riot generate --type hash --index hashIdx

plugins/riot/src/test/resources/generate-json-index

@@ -1 +1 @@
-riot generate --types json --index jsonIdx
+riot generate --type json --index jsonIdx

plugins/riot/src/test/resources/replicate-live-keyslot

@@ -1 +1 @@
-riot replicate --mode liveonly --key-slots 0:8000 redis://source redis://target
+riot replicate --mode liveonly --key-slot 0:8000 redis://source redis://target

plugins/riot/src/test/resources/replicate-live-read-threads

@@ -0,0 +1 @@
+riot replicate --mode live --read-threads 3 redis://source redis://target

plugins/riot/src/test/resources/replicate-no-stream-ids → plugins/riot/src/test/resources/replicate-no-stream-id

@@ -1 +1 @@
-riot replicate --struct --batch 10 redis://source redis://target --no-stream-ids
+riot replicate --struct --batch 10 redis://source redis://target --no-stream-id

plugins/riot/src/test/resources/replicate-no-stream-ids-prune → plugins/riot/src/test/resources/replicate-no-stream-id-prune

@@ -1 +1 @@
-riot replicate --struct --batch 10 redis://source redis://target --no-stream-ids --stream-prune
+riot replicate --struct --batch 10 redis://source redis://target --no-stream-id --stream-prune