Collect successive pairs from a stream

Question

Given a stream such as { 0, 1, 2, 3, 4 },

how can I most elegantly transform it into given form:

{ new Pair(0, 1), new Pair(1, 2), new Pair(2, 3), new Pair(3, 4) }

(assuming, of course, I've defined class Pair)?

Edit: This isn't strictly about ints or primitive streams. The answer should be general for a stream of any type.

Answer 1

This is not elegant, it's a hackish solution, but works for infinite streams

Stream<Pair> pairStream = Stream.iterate(0, (i) -> i + 1).map( // natural numbers
    new Function<Integer, Pair>() {
        Integer previous;

        @Override
        public Pair apply(Integer integer) {
            Pair pair = null;
            if (previous != null) pair = new Pair(previous, integer);
            previous = integer;
            return pair;
        }
    }).skip(1); // drop first null

Now you can limit your stream to the length you want

pairStream.limit(1_000_000).forEach(i -> System.out.println(i));

P.S. I hope there is better solution, something like clojure (partition 2 1 stream)

Answer 2

The operation is essentially stateful so not really what streams are meant to solve - see the "Stateless Behaviors" section in the javadoc:

The best approach is to avoid stateful behavioral parameters to stream operations entirely

One solution here is to introduce state in your stream through an external counter, although it will only work with a sequential stream.

public static void main(String[] args) {
    Stream<String> strings = Stream.of("a", "b", "c", "c");
    AtomicReference<String> previous = new AtomicReference<>();
    List<Pair> collect = strings.map(n -> {
                            String p = previous.getAndSet(n);
                            return p == null ? null : new Pair(p, n);
                        })
                        .filter(p -> p != null)
                        .collect(toList());
    System.out.println(collect);
}


static class Pair<T> {
    private T left, right;
    Pair(T left, T right) { this.left = left; this.right = right; }
    @Override public String toString() { return "{" + left + "," + right + '}'; }
}

Answer 3

You can do this in cyclops-react (I contribute to this library), using the sliding operator.

  LazyFutureStream.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

Or

   ReactiveSeq.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

Assuming the Pair constructor can accept a Collection with 2 elements.

If you wanted to group by 4, and increment by 2 that is also supported.

     ReactiveSeq.rangeLong( 0L,Long.MAX_VALUE)
                .sliding(4,2)
                .forEach(System.out::println);

Equivalant static methods for creating a sliding view over java.util.stream.Stream are also provided in cyclops-streams StreamUtils class.

       StreamUtils.sliding(Stream.of(1,2,3,4),2)
                  .map(Pair::new);

Note :- for single-threaded operation ReactiveSeq would be more appropriate. LazyFutureStream extends ReactiveSeq but is primarily geared for concurrent / parallel use (it is a Stream of Futures).

LazyFutureStream extends ReactiveSeq which extends Seq from the awesome jOOλ (which extends java.util.stream.Stream), so the solutions Lukas' presents would also work with either Stream type. For anyone interested the primary differences between the window / sliding operators are the obvious relative power / complexity trade off and suitability for use with infinite streams (sliding doesn't consume the stream, but buffers as it flows).

Answer 4

I've implemented a spliterator wrapper which takes every n elements T from the original spliterator and produces List<T>:

public class ConsecutiveSpliterator<T> implements Spliterator<List<T>> {

    private final Spliterator<T> wrappedSpliterator;

    private final int n;

    private final Deque<T> deque;

    private final Consumer<T> dequeConsumer;

    public ConsecutiveSpliterator(Spliterator<T> wrappedSpliterator, int n) {
        this.wrappedSpliterator = wrappedSpliterator;
        this.n = n;
        this.deque = new ArrayDeque<>();
        this.dequeConsumer = deque::addLast;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<T>> action) {
        deque.pollFirst();
        fillDeque();
        if (deque.size() == n) {
            List<T> list = new ArrayList<>(deque);
            action.accept(list);
            return true;
        } else {
            return false;
        }
    }

    private void fillDeque() {
        while (deque.size() < n && wrappedSpliterator.tryAdvance(dequeConsumer))
            ;
    }

    @Override
    public Spliterator<List<T>> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return wrappedSpliterator.estimateSize();
    }

    @Override
    public int characteristics() {
        return wrappedSpliterator.characteristics();
    }
}

Following method may be used to create a consecutive stream:

public <E> Stream<List<E>> consecutiveStream(Stream<E> stream, int n) {
    Spliterator<E> spliterator = stream.spliterator();
    Spliterator<List<E>> wrapper = new ConsecutiveSpliterator<>(spliterator, n);
    return StreamSupport.stream(wrapper, false);
}

Sample usage:

consecutiveStream(Stream.of(0, 1, 2, 3, 4, 5), 2)
    .map(list -> new Pair(list.get(0), list.get(1)))
    .forEach(System.out::println);

Answer 5

An elegant solution would be to use zip. Something like:

List<Integer> input = Arrays.asList(0, 1, 2, 3, 4);
Stream<Pair> pairStream = Streams.zip(input.stream(),
                                      input.stream().substream(1),
                                      (a, b) -> new Pair(a, b)
);

This is pretty concise and elegant, however it uses a list as an input. An infinite stream source cannot be processed this way.

Another (lot more troublesome) issue is that zip together with the entire Streams class has been lately removed from the API. The above code only works with b95 or older releases. So with the latest JDK I would say there is no elegant FP style solution and right now we can just hope that in some way zip will be reintroduced to the API.

Answer 6

For calculating successive differences in the time (x-values) of a time-series, I use the stream's collect(...) method:

final List< Long > intervals = timeSeries.data().stream()
                    .map( TimeSeries.Datum::x )
                    .collect( DifferenceCollector::new, DifferenceCollector::accept, DifferenceCollector::combine )
                    .intervals();

Where the DifferenceCollector is something like this:

public class DifferenceCollector implements LongConsumer
{
    private final List< Long > intervals = new ArrayList<>();
    private Long lastTime;

    @Override
    public void accept( final long time )
    {
        if( Objects.isNull( lastTime ) )
        {
            lastTime = time;
        }
        else
        {
            intervals.add( time - lastTime );
            lastTime = time;
        }
    }

    public void combine( final DifferenceCollector other )
    {
        intervals.addAll( other.intervals );
        lastTime = other.lastTime;
    }

    public List< Long > intervals()
    {
        return intervals;
    }
}

You could probably modify this to suit your needs.