Rx - 将数字列表减少为一系列 (number,count)

Question

假设我们有一个数字序列：

0,0,0,0,1,1,2,2,2,3,4,4,4,0,0,0,0,1,1

我正在寻找可以将此序列减少为包含 (element, group_count_of_this_element) 的对的 Rx observable。所以上面会变成： (0,4) // first 4 "0" occurred (1,2) // then there were two "1"s (2,3) // then came three "2"s, one after another (3,1) // and one occurrence of "3" (4,3) // three "4" (0,4) // four zeros (2,1) // and two "1"s

我还在学习，还不能真正掌握这一点。我怀疑 collect 可能会有所帮助，但甚至找不到一个很好的例子......

Answer 1

你可以通过 publish 和 buffer 使用 Observable 边界来实现效果，但我觉得它太复杂了。我发现编写一个将数据和缓冲区更改决策放在一个地方的运算符要简单得多：

public final class BufferUntilChanged<T, U> 
implements Operator<List<T>, T> {
    final Func1<? super T, U> keySelector;

    public BufferUntilChanged(Func1<? super T, U> keySelector) {
        this.keySelector = keySelector;
    }

    @Override
    public Subscriber<? super T> call(Subscriber<? super List<T>> t) {
        BufferUntilChangedSubscriber<T, U> parent = 
                new BufferUntilChangedSubscriber<>(t, keySelector);
        t.add(parent);
        return parent;
    }

    static final class BufferUntilChangedSubscriber<T, U> 
    extends Subscriber<T> {
        final Func1<? super T, U> keySelector;
        final Subscriber<? super List<T>> actual;

        List<T> list;
        U lastKey;

        public BufferUntilChangedSubscriber(
                Subscriber<? super List<T>> actual,
                Func1<? super T, U> keySelector) {
            this.keySelector = keySelector;
            this.actual = actual;
        }

        @Override
        public void onNext(T t) {
            U u;

            try {
                u = keySelector.call(t);
            } catch (Throwable e) {
                unsubscribe();
                actual.onError(e);
                return;
            }

            boolean doRequest;
            if (list == null) {
                list = new ArrayList<>();
                lastKey = u;
                doRequest = true;
            } else
            if (!Objects.equals(lastKey, u)) {
                actual.onNext(list);
                list = new ArrayList<>();
                doRequest = false;
            } else {
                doRequest = true;
            }
            list.add(t);

            lastKey = u;

            if (doRequest) {
                request(1);
            }
        }

        @Override
        public void onError(Throwable e) {
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            if (list != null) {
                actual.onNext(list);
            }
            actual.onCompleted();
        }
    }

}

这是一个使用示例：

    public static void main(String[] args) {
        Observable<Integer> source = Observable.from(
                new Integer[] { 
                        0,0,0,0,
                        1,1,
                        2,2,2,
                        3,
                        4,4,4,
                        0,0,0,0,
                        1,1 });

        source.lift(new BufferUntilChanged<>(v -> v))
        .map(list -> new Integer[] { 
                list.get(0), list.size() 
        })
        .subscribe(v -> 
            System.out.println(Arrays.toString(v)));
    }