resolve #77

assira.core/src/main/java/net/ranides/assira/collection/iterators/SpliteratorUtils.java

@@ -6,6 +6,7 @@
  */
 package net.ranides.assira.collection.iterators;
 
+import java.util.concurrent.atomic.AtomicBoolean;
 import lombok.experimental.UtilityClass;
 
 import net.ranides.assira.functional.Functions;
@@ -28,11 +29,7 @@ public final class SpliteratorUtils {
     /**
      * Returns limited stream, which stops processing at first unmatched element.
      *
-     * Preserves stream parallelization.
-     *
-     * Internally it uses limited spliterator.
-     * Please note, that for parallel streams order of elements and instant termination is not guaranteed.
-     * That means: some (in worst case: all) elements AFTER rejected value could be processed by parallel threads.
+     * Preserves parallelization of all intermediate pipeline steps except "limit" operation itself.
      *
      * @param stream stream
      * @param predicate predicate
@@ -71,31 +68,19 @@ public final class SpliteratorUtils {
         private final Spliterator<T> spliterator;
         private final Predicate<? super T> condition;
         boolean active;
-        Consumer<? super T> current;
-        Consumer<T> adapter;
 
         public LimitSpliterator(Spliterator<T> spliterator, Predicate<? super T> condition) {
             super(spliterator.estimateSize(), spliterator.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED));
             this.spliterator = spliterator;
             this.condition = condition;
-            active = true;
-            adapter = t -> {
-                if ((active = condition.test(t))) { current.accept(t); }
-            };
+            this.active = true;
         }
 
         @Override
         public boolean tryAdvance(Consumer<? super T> action) {
-            if(!active) {
-                return false;
-            }
-            current = action;
-            try {
-                return spliterator.tryAdvance(adapter) && active;
-            }
-            finally {
-                current = null;
-            }
+            return active
+                && spliterator.tryAdvance(t -> { if ((active = condition.test(t))) { action.accept(t); }})
+                && active;
         }
     }
 

assira.core/src/test/java/net/ranides/assira/collection/iterators/SpliteratorUtilsTest.java

@@ -1,5 +1,14 @@
 package net.ranides.assira.collection.iterators;
 
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import net.ranides.assira.collection.lists.IntRange;
+import org.junit.Assert;
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -8,9 +17,10 @@ import net.ranides.assira.collection.query.support.BaseSpliterator;
 import java.util.Arrays;
 import java.util.List;
 
-@Ignore("manual")
+
 public class SpliteratorUtilsTest {
 
+    @Ignore("manual")
     @Test
     public void forEach1() {
         List<Integer> list = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16);
@@ -20,6 +30,7 @@ public class SpliteratorUtilsTest {
         });
     }
 
+    @Ignore("manual")
     @Test
     public void whileEach1() {
         List<Integer> list = Arrays.asList(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32);
@@ -29,4 +40,43 @@ public class SpliteratorUtilsTest {
             return i != 1 && i!=9 && i!=17;
         });
     }
+
+    @Ignore("manual")
+    @Test
+    public void limitThreads() {
+        Set<Integer> set = new HashSet<>(new IntRange(1, 100));
+        SpliteratorUtils.limit(set.parallelStream(), v -> v < 50).forEach(v -> {
+            System.err.println(Thread.currentThread().getName() + "   "+ v);
+        });
+    }
+
+    @Test
+    public void limit() {
+        ArrayList<Integer> list = new ArrayList<>(new IntRange(1, 100));
+        Set<Integer> set = new HashSet<>(new IntRange(1, 100));
+
+        ConcurrentHashMap<Integer, Boolean> out1 = new ConcurrentHashMap<>();
+        SpliteratorUtils.limit(list.stream(), v -> v != 50).forEach(v -> out1.put(v, true));
+
+        ConcurrentHashMap<Integer, Boolean> out2 = new ConcurrentHashMap<>();
+        SpliteratorUtils.limit(list.stream(), v -> v != 50).forEach(v -> out2.put(v, true));
+
+        ConcurrentHashMap<Integer, Boolean> out3 = new ConcurrentHashMap<>();
+        SpliteratorUtils.limit(set.parallelStream(), v -> v != 50).forEach(v -> out3.put(v, true));
+
+        Set<Integer> out4 = SpliteratorUtils.limit(set.parallelStream(), v -> v != 50).collect(Collectors.toSet());
+
+        Set<Integer> out5 = SpliteratorUtils.limit(IntStream.rangeClosed(1, 100).boxed(), v -> v != 50).collect(Collectors.toSet());
+
+        Set<Integer> out6 = SpliteratorUtils.limit(IntStream.rangeClosed(1, 100).boxed().parallel(), v -> v != 50).collect(Collectors.toSet());
+
+        Set<Integer> expected = new TreeSet<>(new IntRange(1, 50));
+
+        Assert.assertEquals(expected, new TreeSet<>(out1.keySet()));
+        Assert.assertEquals(expected, new TreeSet<>(out2.keySet()));
+        Assert.assertEquals(expected, new TreeSet<>(out3.keySet()));
+        Assert.assertEquals(expected, out4);
+        Assert.assertEquals(expected, out5);
+        Assert.assertEquals(expected, out6);
+    }
 }