Fix Streamable range traversal starts and maximum endpoints (#8337)

Use the same index-based step for asynchronous and enumerable consumption. Comparing the next index to the exclusive endpoint supports ranges ending at the primitive maximum without changing indexed collection.

Generated-by: OpenAI Codex

Author: PHJ2000

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRange.java

@@ -36,18 +36,17 @@ static final class RangeStreamer<T>
 
         int current;
 
+        int index;
+
         RangeStreamer(int start, int end) {
             this.start = start;
-            this.current = start - 1;
+            this.index = start;
             this.end = end;
         }
 
         @Override
         public @NonNull CompletionStage<Boolean> next() {
-            if (++current >= end) {
-                return NEXT_FALSE;
-            }
-            return NEXT_TRUE;
+            return nextSync() ? NEXT_TRUE : NEXT_FALSE;
         }
 
         @Override
@@ -57,6 +56,7 @@ static final class RangeStreamer<T>
 
         @Override
         public @NonNull CompletionStage<Void> finish() {
+            index = end;
             current = end;
             return FINISHED;
         }
@@ -72,10 +72,13 @@ public long limit() {
         }
 
         @Override
-        public boolean nextSync() throws Throwable {
-            if (++current >= end) {
+        public boolean nextSync() {
+            int i = index;
+            if (i == end) {
                 return false;
             }
+            current = i;
+            index = i + 1;
             return true;
         }
     }

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeLong.java

@@ -46,13 +46,7 @@ static final class RangeLongStreamer<T>
 
         @Override
         public @NonNull CompletionStage<Boolean> next() {
-            long i = index;
-            if (i >= end) {
-                return NEXT_FALSE;
-            }
-            current = i;
-            index = i + 1;
-            return NEXT_TRUE;
+            return nextSync() ? NEXT_TRUE : NEXT_FALSE;
         }
 
         @Override
@@ -78,10 +72,13 @@ public long limit() {
         }
 
         @Override
-        public boolean nextSync() throws Throwable {
-            if (++current >= end) {
+        public boolean nextSync() {
+            long i = index;
+            if (i == end) {
                 return false;
             }
+            current = i;
+            index = i + 1;
             return true;
         }
     }

src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableRangeTest.java

@@ -124,4 +124,96 @@ public void longEnumerated() throws Throwable {
         .awaitDone(5, TimeUnit.SECONDS)
         .assertResult(List.of(1L, 2L, 3L));
     }
+
+    @Test
+    public void maximum() throws Throwable {
+        Streamable.range(Integer.MAX_VALUE - 1, 2)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(Integer.MAX_VALUE - 1, Integer.MAX_VALUE);
+    }
+
+    @Test
+    public void maximumLong() throws Throwable {
+        Streamable.rangeLong(Long.MAX_VALUE - 1, 2)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(Long.MAX_VALUE - 1, Long.MAX_VALUE);
+    }
+
+    @Test
+    public void minimum() throws Throwable {
+        Streamable.range(Integer.MIN_VALUE, 2)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(Integer.MIN_VALUE, Integer.MIN_VALUE + 1);
+    }
+
+    @Test
+    public void minimumLong() throws Throwable {
+        Streamable.rangeLong(Long.MIN_VALUE, 2)
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(Long.MIN_VALUE, Long.MIN_VALUE + 1);
+    }
+
+    @Test
+    public void maximumEnumerated() throws Throwable {
+        Streamable.range(Integer.MAX_VALUE - 1, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(Integer.MAX_VALUE - 1, Integer.MAX_VALUE));
+    }
+
+    @Test
+    public void maximumEnumeratedLong() throws Throwable {
+        Streamable.rangeLong(Long.MAX_VALUE - 1, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(Long.MAX_VALUE - 1, Long.MAX_VALUE));
+    }
+
+    @Test
+    public void minimumEnumeratedLong() throws Throwable {
+        Streamable.rangeLong(Long.MIN_VALUE, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(Long.MIN_VALUE, Long.MIN_VALUE + 1));
+    }
+
+    @Test
+    public void longEnumeratedFromZero() throws Throwable {
+        Streamable.rangeLong(0, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(0L, 1L));
+    }
+
+    @Test
+    public void longEnumeratedFromNegative() throws Throwable {
+        Streamable.rangeLong(-2, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(-2L, -1L));
+    }
+
+    @Test
+    public void longEnumeratedFromOffset() throws Throwable {
+        Streamable.rangeLong(5, 2)
+        .filter(_ -> true)
+        .collect(Collectors.toList())
+        .test()
+        .awaitDone(5, TimeUnit.SECONDS)
+        .assertResult(List.of(5L, 6L));
+    }
 }