StreamUtil.interleaving() giving strange ClassCastException

[kash74]

Method StreamUtil.interleaving() seems to be broken in v1.14. I keep getting this exception:

java.lang.ClassCastException: [Ljava.lang.Object; cannot be cast to [Ljava.lang.String;
	at com.codepoetics.protonpack.InterleavingSpliterator.tryAdvance(InterleavingSpliterator.java:48)
	at java.util.Spliterator.forEachRemaining(Spliterator.java:326)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:499)
	at com.swheel.core.StreamUtilsTest.interleaving(StreamUtilsTest.java:40)

The following unit test reproduces this:

public class StreamUtilsTest {

    String cn11 = "11";
    String cn12 = "12";
    String cn21 = "21";
    String cn22 = "22";

    List<String> list1 = Arrays.asList(cn11, cn12);
    List<String> list2 = Arrays.asList(cn21, cn22);

    @Test
    public void interleaving() {
        AtomicLong evenOdd = new AtomicLong();
        Function<String[], Integer> alternating = cns -> {
            assert cns.length == 2;
            if (cns[0] == null) return 1;
            if (cns[1] == null) return 0;
            if (evenOdd.getAndIncrement() % 2 == 0) {
                return 0;
            } else {
                return 1;
            }
        };

        List<String> res =
            StreamUtils.interleave(alternating, Arrays.asList(list1.stream(), list2.stream()))
                    .collect(Collectors.toList());

        Assert.assertEquals(Arrays.asList(cn11, cn21, cn12, cn22), res);
    }
}

I get no compile errors for this test, but the strange ClassClast at runtime.

[hadrienk]

The problem here is type erasure I think. If you wrap your lambda in a parameterized method (like it is done in Selectors.roundRobin()) I think it should work.

I am not an expert but I have solved this kind of problem by using actual collections instead of arrays. I can put a patch to illustrate.

[hadrienk]

Index: src/test/java/com/codepoetics/protonpack/InterleaveTest.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- src/test/java/com/codepoetics/protonpack/InterleaveTest.java	(revision 473d6d7beac157ea94ae4a9165ab4cfff87de34e)
+++ src/test/java/com/codepoetics/protonpack/InterleaveTest.java	(revision )
@@ -1,10 +1,12 @@
 package com.codepoetics.protonpack;
 
 import com.codepoetics.protonpack.selectors.Selectors;
+import org.hamcrest.Matchers;
 import org.junit.Test;
 
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -13,7 +15,27 @@
 import static org.hamcrest.Matchers.contains;
 
 public class InterleaveTest {
+    
+    @Test public void
+    lambda_interleaving() {
 
+        Stream<String> stream1 = Stream.of("11", "12");
+        Stream<String> stream2 = Stream.of("21", "22");
+
+        AtomicLong evenOdd = new AtomicLong();
+        Stream<String> interleaved = StreamUtils.interleave(cns -> {
+            if (cns.get(0) == null) return 1;
+            if (cns.get(1) == null) return 0;
+            if (evenOdd.getAndIncrement() % 2 == 0) {
+                return 0;
+            } else {
+                return 1;
+            }
+        }, stream1, stream2);
+
+        assertThat(interleaved.collect(Collectors.toList()), Matchers.contains("11", "21", "12", "22"));
+    }
+
     @Test public void
     round_robin_interleaving() {
         Stream<String> streamA = Stream.of("Peter", "Paul", "Mary");
Index: src/main/java/com/codepoetics/protonpack/InterleavingSpliterator.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- src/main/java/com/codepoetics/protonpack/InterleavingSpliterator.java	(revision 473d6d7beac157ea94ae4a9165ab4cfff87de34e)
+++ src/main/java/com/codepoetics/protonpack/InterleavingSpliterator.java	(revision )
@@ -1,21 +1,24 @@
 package com.codepoetics.protonpack;
 
+import com.codepoetics.protonpack.selectors.Selector;
+
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Spliterator;
 import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
 
 class InterleavingSpliterator<T> implements Spliterator<T> {
 
-    public static <T> Spliterator<T> interleaving(Spliterator<T>[] spliterators, Function<T[], Integer> selector) {
-        Supplier<T[]> bufferedValues = () -> {
-            T[] values = (T[]) new Object[spliterators.length];
+    public static <T> Spliterator<T> interleaving(Spliterator<T>[] spliterators, Selector<T> selector) {
+        Supplier<ArrayList<T>> bufferedValues = () -> {
+            ArrayList<T> values = new ArrayList<>(spliterators.length);
 
             for (int i=0; i<spliterators.length; i++) {
                 final int stableIndex = i;
-                spliterators[i].tryAdvance(t -> values[stableIndex] = t);
+                spliterators[i].tryAdvance(t -> values.add(stableIndex, t));
             }
 
             return values;
@@ -25,11 +28,11 @@
     }
 
     private final Spliterator<T>[] spliterators;
-    private final Supplier<T[]> bufferSupplier;
-    private T[] buffer = null;
-    private final Function<T[], Integer> selector;
+    private final Supplier<ArrayList<T>> bufferSupplier;
+    private List<T> buffer = null;
+    private final Selector<T> selector;
 
-    private InterleavingSpliterator(Spliterator<T>[] spliterators, Supplier<T[]> bufferSupplier, Function<T[], Integer> selector) {
+    private InterleavingSpliterator(Spliterator<T>[] spliterators, Supplier<ArrayList<T>> bufferSupplier, Selector<T> selector) {
         this.spliterators = spliterators;
         this.bufferSupplier = bufferSupplier;
         this.selector = selector;
@@ -41,15 +44,15 @@
             buffer = bufferSupplier.get();
         }
 
-        if (Stream.of(buffer).allMatch(Predicate.isEqual(null))) {
+        if (buffer.stream().allMatch(Predicate.isEqual(null))) {
             return false;
         }
 
         int selected = selector.apply(buffer);
-        action.accept(buffer[selected]);
+        action.accept(buffer.get(selected));
 
-        if (!spliterators[selected].tryAdvance(t -> buffer[selected] = t)) {
-            buffer[selected] = null;
+        if (!spliterators[selected].tryAdvance(t -> buffer.set(selected,t))) {
+            buffer.set(selected, null);
         }
 
         return true;
Index: src/main/java/com/codepoetics/protonpack/selectors/Selector.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- src/main/java/com/codepoetics/protonpack/selectors/Selector.java	(revision 473d6d7beac157ea94ae4a9165ab4cfff87de34e)
+++ src/main/java/com/codepoetics/protonpack/selectors/Selector.java	(revision )
@@ -1,7 +1,8 @@
 package com.codepoetics.protonpack.selectors;
 
+import java.util.List;
 import java.util.function.Function;
 
 @FunctionalInterface
-public interface Selector<T> extends Function<T[], Integer> {
+public interface Selector<T> extends Function<List<T>, Integer> {
 }
Index: src/main/java/com/codepoetics/protonpack/selectors/Selectors.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- src/main/java/com/codepoetics/protonpack/selectors/Selectors.java	(revision 473d6d7beac157ea94ae4a9165ab4cfff87de34e)
+++ src/main/java/com/codepoetics/protonpack/selectors/Selectors.java	(revision )
@@ -1,7 +1,8 @@
 package com.codepoetics.protonpack.selectors;
 
 import java.util.Comparator;
-import java.util.stream.Stream;
+import java.util.List;
+import java.util.Objects;
 
 public final class Selectors {
     private Selectors() {
@@ -11,13 +12,13 @@
         return new Selector<T>() {
             private int startIndex = 0;
             @Override
-            public Integer apply(T[] options) {
+            public Integer apply(List<T> options) {
                 int result = startIndex;
-                while (options[result] == null) {
-                    result = (result + 1) % options.length;
+                while (options.get(result) == null) {
+                    result = (result + 1) % options.size();
                 }
 
-                startIndex = (result + 1) % options.length;
+                startIndex = (result + 1) % options.size();
                 return result;
             }
         };
@@ -33,15 +34,15 @@
             private int startIndex = 0;
 
             @Override
-            public Integer apply(T[] options) {
-                T smallest = Stream.of(options).filter(t -> t != null).min(comparator).get();
+            public Integer apply(List<T> options) {
+                T smallest = options.stream().filter(Objects::nonNull).min(comparator).get();
 
                 int result = startIndex;
-                while (options[result] == null || comparator.compare(smallest, options[result]) != 0) {
-                    result = (result + 1) % options.length;
+                while (options.get(result) == null || comparator.compare(smallest, options.get(result)) != 0) {
+                    result = (result + 1) % options.size();
                 }
 
-                startIndex = (result + 1) % options.length;
+                startIndex = (result + 1) % options.size();
                 return result;
             }
         };
Index: src/main/java/com/codepoetics/protonpack/StreamUtils.java
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
--- src/main/java/com/codepoetics/protonpack/StreamUtils.java	(revision 473d6d7beac157ea94ae4a9165ab4cfff87de34e)
+++ src/main/java/com/codepoetics/protonpack/StreamUtils.java	(revision )
@@ -1,6 +1,7 @@
 package com.codepoetics.protonpack;
 
 import com.codepoetics.protonpack.functions.TriFunction;
+import com.codepoetics.protonpack.selectors.Selector;
 
 import java.util.*;
 import java.util.function.*;
@@ -337,7 +338,7 @@
      * @param <T> The type over which the interleaved streams stream.
      * @return An interleaved stream.
      */
-    public static <T> Stream<T> interleave(Function<T[], Integer> selector, Stream<T>... streams) {
+    public static <T> Stream<T> interleave(Selector<T> selector, Stream<T>... streams) {
         Spliterator<T>[] spliterators = (Spliterator<T>[]) Stream.of(streams).map(s -> s.spliterator()).toArray(Spliterator[]::new);
         return StreamSupport.stream(InterleavingSpliterator.interleaving(spliterators, selector), false)
                 .onClose(closerFor(streams));
@@ -357,7 +358,7 @@
      * @param <T> The type over which the interleaved streams stream.
      * @return An interleaved stream.
      */
-    public static <T> Stream<T> interleave(Function<T[], Integer> selector, List<Stream<T>> streams) {
+    public static <T> Stream<T> interleave(Selector<T> selector, List<Stream<T>> streams) {
         Spliterator<T>[] spliterators = (Spliterator<T>[]) streams.stream().map(s -> s.spliterator()).toArray(Spliterator[]::new);
         return StreamSupport.stream(InterleavingSpliterator.interleaving(spliterators, selector), false)
                 .onClose(closerFor(streams));

[hadrienk]

This does it. I think the Supplier<ArrayList<T>> bufferedValues could probably be taken in in the InterleavingSpliterator as a method IMO.

[poetix]

It's a (very peculiar) peculiarity of type erasure. I still don't quite understand why Selectors.roundRobin() works, while pulling the selector created in Selectors.roundRobin() inline like so:

@Test
  public void interleaving() {
    Selector<String> alternating = new Selector<String>() {
      private int startIndex = 0;
      @Override
      public Integer apply(String[] options) {
        int result = startIndex;
        while (options[result] == null) {
          result = (result + 1) % options.length;
        }

        startIndex = (result + 1) % options.length;
        return result;
      }
    };

    List<String> res =
        StreamUtils.interleave(alternating, Arrays.asList(list1.stream(), list2.stream()))
            .collect(Collectors.toList());

    Assert.assertEquals(Arrays.asList(cn11, cn21, cn12, cn22), res);
  }

fails with the ClassCastException. Something to do with the binding of type variables? It's a bit of a corner case (from the point of view of the type checker), admittedly.

Shifting to generic collections instead of arrays would fix this, but it would also break any user-written selectors that worked with arrays. I'm also not so keen on paying the allocation cost of a new ArrayList for every tryAdvance() - creating a single ArrayList up-front and mutating its contents would be more efficient. Better still might be to create an object representing the available selections -

interface Selectable<T> {
  int range(); // the range of indices to select from
  boolean hasValue(int index); // whether there is a value at the given index
  T getValue(int index); // get the value at the given index (or throw some exception if there isn't one)
}

and having Selectors work with that instead, eliminating nulls from the client contract. Any thoughts?

[poetix]

(the answer to "why does this happen" is that the generic <T> Selector<T> roundRobin() method actually creates, after erasure of T, a Selector<Object>, which expects an array of Object[] and doesn't complain when that's what it's passed. In this case, it doesn't pay to be too concrete about what array element type you expect!)

[hadrienk]

The answer to "why does this happen" is that the generic <T> Selector<T> roundRobin() method actually creates, after erasure of T, a Selector<Object>, which expects an array of Object[] and doesn't complain when that's what it's passed. In this case, it doesn't pay to be too concrete about what array element type you expect!

I was under the impression that the buffer creation in a lambda (Supplier<T[]> bufferedValues = () -> {) was part of the problem. But I am a bit lost here.

Shifting to generic collections instead of arrays would fix this, but it would also break any user-written selectors that worked with arrays. I'm also not so keen on paying the allocation cost of a new ArrayList for every tryAdvance() - creating a single ArrayList up-front and mutating its contents would be more efficient. Better still might be to create an object representing the available selections.

Absolutely, creating a new ArrayList is a no go. Wrapping the array with Arrays.asList() is also an option?

eliminating nulls from the client contract. Any thoughts?

Using an interface is a good idea. Maybe Optional as a return value to communicate the intent better? But this would break already created selector unless we support both.