Find Sequence in Observable.

rxjava_rollingBuffer.java

@Test
    public void rollingBufferSearchForSequenceTest() throws Exception {
        TestObserver<List<Integer>> listTestObserver = Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5, 6, 1, 2, 3, 4, 5, 6))
                .compose(searchForSequence(Arrays.asList(1, 2)))
                .test()
                .assertComplete()
                .assertValueCount(2);

        List<Integer> collect = listTestObserver.values()
                .stream()
                .flatMap(Collection::stream)
                .collect(Collectors.toList());

        assertThat(collect, contains(1, 2, 1, 2));
    }

    private static <T> ObservableTransformer<T, List<T>> searchForSequence(Collection<T> items) {
        return upstream -> upstream.buffer(items.size(), 1)
                .filter(integers -> integers.size() == items.size())
                .filter(integers -> integers.containsAll(items));
    }