Turning an array of Java8 threads into a tuple stream

Question

Let's say I have an array of Java 8 threads: Stream<T>[] streamsI would like to create a Stream, where each element of the new stream is an array consisting of selecting one element from each of the initial base threads (let's assume that they are all sequential).

For example, if I have:

  streams [ 0 ] returning: ( "A", "B", "C" ), 
  streams [ 1 ] returning ( "X", "Y", "Z" ) 
  and streams [ 2 ] as ( "0", "1", "2" )

I need a stream that returns

  ( { "A", "X", "0" }, { "B", "Y", "1" }, { "C", "Z", "2" } )

Is there any code that already implements this? I have an idea how to do this, this will be a generalization of a pair of pairs, but I would like to know if there is something reusable.

EDIT : sorry, I realized that I need clarification:

• I don’t want to create the whole matrix, I want the thread to dynamically return one row at a time (first A / X / 0, then B / Y / 1, etc.), without having to occupy the memory with all the rows in advance. I am fine with reasonable assumptions regarding the size of the underlying threads (for example, taking a minimum, stopping as soon as there is a thread that has no more items to return).

• I know that this can be realized by first turning the underlying threads into iterators, and then creating a new iterator, which from the next () selects one element from each of the underscoring iterators and returns a new line. This is what the couple example that I linked above does, and I could implement it this way on myself, here I am trying to figure out if this has already been done in some library (I know that the JDK does not have such a function).

Answer 1

, , , :

• TupleSpliterator, , spliterators;
• Tuple Stream Builder, , .
• Spliteraror/Iterator parallelism ( ), - , , TupleIterator.

, ( ), .

EDIT. Spliterator Federico, , Iterator .

Answer 2

-, , .

, JDK. zip, Tuples, , , :

Stream[] streams = Stream.of(
  Stream.of("A", "B", "C"),
  Stream.of("X", "Y", "Z"),
  Stream.of("0", "1", "2"))
    .toArray(Stream[]::new);

String[][] arrays = Arrays.stream(streams)
  .map(s -> s.toArray(String[]::new))
  .toArray(String[][]::new);

int minSize = Arrays.stream(arrays)
  .mapToInt(s -> s.length)
  .min().orElse(0);

String[][] zipped = IntStream.range(0, minSize)
  .mapToObj(i -> Arrays.stream(arrays)
  .map(s -> s[i])
    .toArray(String[]::new))
  .toArray(String[][]::new);

-, - , .

-, , , , zip, , .

-, zipping (IntStream.range(0, minSize)) .

.get() on , minSize , - .

, , :

List<List<String>> lists = Arrays.asList(
  Arrays.asList("A", "B", "C"),
  Arrays.asList("X", "Y", "Z"),
  Arrays.asList("0", "1", "2"));

final int minSize = lists.stream()
  .mapToInt(List::size)
  .min().orElse(0);

List<List<String>> result = IntStream.range(0, minSize)
  .mapToObj(i -> lists.stream()
  .map(s -> s.get(i))
    .collect(Collectors.toList()))
  .collect(Collectors.toList());

API Java 9 Stream API, , minSize.

, lazy, :

IntStream.range(0, minSize)
  .mapToObj(i -> lists.stream()
    .map(s -> s.get(i))
    .collect(Collectors.toList()));

Answer 3

Stream - TupleX, , , ( ), :

@SafeVarargs
static <T> Stream<Stream<T>> streamOfStreams(Stream<T>... streams) {

    @SuppressWarnings("unchecked")
    Iterator<T>[] iterators = new Iterator[streams.length];
    for (int i = 0; i < streams.length; ++i) {
        iterators[i] = streams[i].iterator();
    }

    Iterator<T> first = iterators[0];

    Builder<Stream<T>> outer = Stream.builder();
    Builder<T> inner = Stream.builder();
    while (first.hasNext()) {
        for (int i = 0; i < streams.length; ++i) {
            inner.add(iterators[i].next());
        }
        outer.add(inner.build());
        inner = Stream.builder();
    }

    return outer.build();
}

Answer 4

Guava 21 Streams.zip, , , , .

, , Streams.zip :

Stream<List<String>> zipped = Arrays.stream(streams)
    .map(s -> s.map(e -> {
        List<String> l = new ArrayList<>();
        l.add(e);
        return l;
    }))
    .reduce((s1, s2) -> Streams.zip(s1, s2, (l1, l2) -> {
        l1.addAll(l2);
        return l1;
    }))
    .orElse(Stream.empty());

List<List<String>> tuples = zipped.collect(Collectors.toList());

System.out.println(tuples); // [[A, X, 0], [B, Y, 1], [C, Z, 2]]

, Stream<T> Stream<List<T>>, List.addAll zip-.

---

: , , - .

, Stream.reduce, , :

Stream<List<String>> zipped = Arrays.stream(streams)
    .reduce(
        IntStream.range(0, streams.length).mapToObj(n -> new ArrayList<>()),
        (z, s) -> Streams.zip(z, s, (l, e) -> {
            l.add(e);
            return l;
        }),
        (s1, s2) -> Streams.zip(s1, s2, (l1, l2) -> {
            l1.addAll(l2);
            return l1;
        }));

List<List<String>> tuples = zipped.collect(Collectors.toList());

System.out.println(tuples); // [[A, X, 0], [B, Y, 1], [C, Z, 2]]

n , n streams, Streams.zip zip . , : Streams.zip .