According to Rx contract, I think all operators should support the following observer.

```java
        Observer<Integer> o = new Observer<Integer>() {
            private int count = 0;

            @Override
            public void onCompleted() {
                count++;
            }

            @Override
            public void onError(Throwable e) {
                count++;
            }

            @Override
            public void onNext(Integer args) {
                count++;
            }
        };
```