How to use Java 8 Stream::collect() to find the longest sequential subsequence in a stream

readme.md

Prompted by the question here on stack overflow, this is how to use a collector to find the longest subsequence of consecutive integers in a stream.

It seems to be a lot more work than a typical loop over a list/array, but on the other hand it will work on a stream of indefinite length in linear time with storage linear in the size of the longest sequential subsequence.

And it also seems to be a good reason to investigate the StreamEx library, 'cause it has helpers for this class of problem (along with much else).

UsingStreamCollectOrCollector.java

package com.bakins_bits;

import static org.assertj.core.api.Assertions.assertThat;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;

import org.testng.annotations.Test;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;

public class UsingStreamCollectOrCollector
{
    @Test
    public void longest_sequential_subsequence_in_sequence_using_collect() {
        // ARRANGE
        List<Integer> testSequence = ImmutableList.of(1, 3, 4, 5, 5, -4, 7, 9, 10, 11, 12, 14, 14, 14, 14, 14, -6, -5,
            -4, 0);

        // ACT
        class R
        {
            List<Integer> current = new ArrayList<Integer>();
            List<Integer> longest = new ArrayList<Integer>();

            public List<Integer> getResult() {
                return (current.size() > longest.size()) ? current : longest;
            }
        }

        BiConsumer<R, Integer> accumulator = (r, i) -> {
            if (r.current.isEmpty())
                r.current.add(i);
            else if (1 + r.current.get(r.current.size() - 1) == i)
                r.current.add(i);
            else {
                if (r.longest.size() < r.current.size()) {
                    r.longest = r.current;
                    r.current = new ArrayList<Integer>();
                } else
                    r.current.clear();
                r.current.add(i);
            }
        };

        BiConsumer<R, R> combiner = (__, ___) -> {
        };

        List<Integer> actual = testSequence.stream().sequential().collect(R::new, accumulator, combiner).getResult();

        // ASSERT
        List<Integer> expected = ImmutableList.of(9, 10, 11, 12);
        assertThat(actual).containsExactlyElementsOf(expected);
    }

    @Test
    public void longest_sequential_subsequence_in_sequence_using_collector() {
        // ARRANGE
        List<Integer> testSequence = ImmutableList.of(1, 3, 4, 5, 5, -4, 7, 9, 10, 11, 12, 14, 14, 14, 14, 14, -6, -5,
            -4, 0);

        // ACT
        class R implements Collector<Integer, R, List<Integer>>
        {
            List<Integer> current = new ArrayList<Integer>();
            List<Integer> longest = new ArrayList<Integer>();

            @Override
            public Supplier<R> supplier() {
                return () -> new R();
            }

            @Override
            public BiConsumer<R, Integer> accumulator() {
                return (r, i) -> {
                    if (r.current.isEmpty())
                        r.current.add(i);
                    else if (1 + r.current.get(r.current.size() - 1) == i)
                        r.current.add(i);
                    else {
                        if (r.longest.size() < r.current.size()) {
                            r.longest = r.current;
                            r.current = new ArrayList<Integer>();
                        } else
                            r.current.clear();
                        r.current.add(i);
                    }
                };
            }

            @Override
            public BinaryOperator<R> combiner() {
                return (__, ___) -> __;
            }

            @Override
            public Function<R, List<Integer>> finisher() {
                return r -> r.current.size() > r.longest.size() ? r.current : r.longest;
            }

            @Override
            public Set<R.Characteristics> characteristics() {
                return ImmutableSet.of();
            }
        }

        List<Integer> actual = testSequence.stream().sequential().collect(new R());

        // ASSERT
        List<Integer> expected = ImmutableList.of(9, 10, 11, 12);
        assertThat(actual).containsExactlyElementsOf(expected);
    }
}