RxJava - using a custom stateful Operator

IndexOperator.java

// Maps an emitted item to an Integer designating the item's zero-based index
public class IndexOperator<T> implements Observable.Operator<Integer, T> {

	private int counter;

	public IndexOperator() {
		counter = 0;
	}

	@Override
	public Subscriber<? super T> call(final Subscriber<? super Integer> subscriber) {
		return new Subscriber<T>() {
			@Override
			public void onCompleted() {
				if (!subscriber.isUnsubscribed()) {
					subscriber.onCompleted();
				}
			}

			@Override
			public void onError(Throwable e) {
				if (!subscriber.isUnsubscribed()) {
					subscriber.onError(e);
				}
			}

			@Override
			public void onNext(T t) {
				if (!subscriber.isUnsubscribed()) {
					subscriber.onNext(counter);
					counter++;
				}
			}
		};
	}

}

Usage

Observable<Foo> emitter = ...;
emitter.lift(new IndexOperator<Foo>).subscribe(new Subscriber<Integer>() {...});