use CursorId in ScanIteration to avoid long overflow

- Spring Data 3.3 deprecated `long cursorId` in favor of String-backed `CursorId` object to avoid `NumberFormatException` when then cursor ID is bigger than max long
 - more info: https://github.com/spring-projects/spring-data-redis/issues/2796

Signed-off-by: Vlastimil Kotas <vlastimil.kotas@jumio.com>

redisson-spring-data/redisson-spring-data-33/src/main/java/org/redisson/spring/data/connection/RedissonClusterConnection.java

@@ -389,7 +389,7 @@ public class RedissonClusterConnection extends RedissonConnection implements Red
             private RedisClient client = getEntry(node);
             
             @Override
-            protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+            protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
                 if (isQueueing() || isPipelined()) {
                     throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
                 }
@@ -399,7 +399,7 @@ public class RedissonClusterConnection extends RedissonConnection implements Red
                 }
                 
                 List<Object> args = new ArrayList<Object>();
-                args.add(Long.toUnsignedString(cursorId));
+                args.add(cursorId);
                 if (options.getPattern() != null) {
                     args.add("MATCH");
                     args.add(options.getPattern());
@@ -413,11 +413,11 @@ public class RedissonClusterConnection extends RedissonConnection implements Red
                 ListScanResult<byte[]> res = syncFuture(f);
                 String pos = res.getPos();
                 client = res.getRedisClient();
-                if ("0".equals(pos)) {
+                if (CursorId.isInitial(pos)) {
                     client = null;
                 }
                 
-                return new ScanIteration<byte[]>(Long.parseUnsignedLong(pos), res.getValues());
+                return new ScanIteration<byte[]>(CursorId.of(pos), res.getValues());
             }
         }.open();
     }

redisson-spring-data/redisson-spring-data-33/src/main/java/org/redisson/spring/data/connection/RedissonConnection.java

@@ -252,14 +252,14 @@ public class RedissonConnection extends AbstractRedisConnection {
 
     @Override
     public Cursor<byte[]> scan(ScanOptions options) {
-        return new ScanCursor<byte[]>(0, options) {
+        return new ScanCursor<byte[]>(Cursor.CursorId.initial(), options) {
 
             private RedisClient client;
             private Iterator<MasterSlaveEntry> entries = redisson.getConnectionManager().getEntrySet().iterator();
             private MasterSlaveEntry entry = entries.next();
             
             @Override
-            protected ScanIteration<byte[]> doScan(long cursorId, ScanOptions options) {
+            protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
                 if (isQueueing() || isPipelined()) {
                     throw new UnsupportedOperationException("'SSCAN' cannot be called in pipeline / transaction mode.");
                 }
@@ -269,10 +269,10 @@ public class RedissonConnection extends AbstractRedisConnection {
                 }
                 
                 List<Object> args = new ArrayList<Object>();
-                if (cursorId == 101010101010101010L) {
+                if (CursorId.of("101010101010101010").equals(cursorId)) {
-                    cursorId = 0;
+                    cursorId = CursorId.initial();
                 }
-                args.add(Long.toUnsignedString(cursorId));
+                args.add(cursorId);
                 if (options.getPattern() != null) {
                     args.add("MATCH");
                     args.add(options.getPattern());
@@ -286,7 +286,7 @@ public class RedissonConnection extends AbstractRedisConnection {
                 ListScanResult<byte[]> res = syncFuture(f);
                 String pos = res.getPos();
                 client = res.getRedisClient();
-                if ("0".equals(pos)) {
+                if (CursorId.isInitial(pos)) {
                     if (entries.hasNext()) {
                         pos = "101010101010101010";
                         entry = entries.next();
@@ -296,7 +296,7 @@ public class RedissonConnection extends AbstractRedisConnection {
                     }
                 }
                 
-                return new ScanIteration<byte[]>(Long.parseUnsignedLong(pos), res.getValues());
+                return new ScanIteration<byte[]>(CursorId.of(pos), res.getValues());
             }
         }.open();
     }