RXJava PausableBuffer

Question

Hi I'm a newbie with RXJava. I looking for an observable solution that would continue and pause emitting items according to what items are received.

Lets say our condition is this Integer Predicate:

Func1<Integer, Boolean> isOdd = number -> number%2==1;

When we add numbers to Subject like : myNumberSubject.onNext(someInt); all the numbers that are added before an odd number is added are stocked in a buffer, but ones the first odd number is added, all the numbers in the buffer are emitted in one burst (including the item that was odd).

After that every number is emitted one by one as long as it is odd. Ones again when an Even number is added it is put to the buffer.

I haven been able to find any real working examples but this marbles example of pausableBuffer could have the potential to do exactly what I am trying to do. http://rxmarbles.com/#pausableBuffered I hope there is some pre-existing RXJava solution that would do the trick. Here is my own hack job solution.

public class PausableBuffer<R> {

    private boolean isPaused;
    private List<R> buffer;
    private ReplaySubject<R> regulatedSubject;

    private PausableBuffer(){
        regulatedSubject = ReplaySubject.create();
        buffer=new LinkedList<>();
    }

    public static <R>Observable<R> create(Observable<R> observable, Func1<R, Boolean> continueCondition){
        PausableBuffer<R> pausableBuffer = new PausableBuffer<>();
        observable.subscribe(value -> {
            synchronized(pausableBuffer) {
                if(pausableBuffer.isPaused){
                    pausableBuffer.buffer.add(value);
                    if(continueCondition.call(value)){
                        pausableBuffer.isPaused=false;
                        for (R r : pausableBuffer.buffer) {
                            pausableBuffer.regulatedSubject.onNext(r);
                        }
                        pausableBuffer.buffer.clear();
                    }
                }else{
                    if(continueCondition.call(value)){
                        pausableBuffer.regulatedSubject.onNext(value);
                    }else{
                        pausableBuffer.isPaused=true;
                        pausableBuffer.buffer.add(value);
                    }
                }
            }
        });
        return pausableBuffer.regulatedSubject.asObservable();
    }

    public static void main(String[] args) {
        BehaviorSubject<Integer> behaviorSubject = BehaviorSubject.create();
        Observable<Integer> observable = PausableBuffer.<Integer>create(
                behaviorSubject.asObservable(),                    
                intValue -> intValue==5 || 6<intValue);//continueCondition
        observable.subscribe(v -> System.out.print(v+", "));
        for (int i = 0; i <= 8; i++) {
            System.out.print("adding " + i + " : ");
            behaviorSubject.onNext(i);
            System.out.println();
        }
    }
}

prints out:

• adding 0 :
• adding 1 :
• adding 2 :
• adding 3 :
• adding 4 :
• adding 5 : 0, 1, 2, 3, 4, 5,
• adding 6 :
• adding 7 : 6, 7,
• adding 8 : 8,

Answer 1

Second version:

public final class ContinueWhile<T> implements Observable.Operator<T, T> {

final Func1<T, Boolean> continuePredicate;

private ContinueWhile(Func1<T, Boolean> continuePredicate) {
    this.continuePredicate = continuePredicate;
}

public static <T>ContinueWhile<T> create(Func1<T, Boolean> whileTrue){
    return new ContinueWhile<>(whileTrue);
}

@Override
public Subscriber<? super T> call(Subscriber<? super T> child) {
    ContinueWhileSubscriber parent = new ContinueWhileSubscriber(child);
    child.add(parent);
    return parent;
}

final class ContinueWhileSubscriber extends Subscriber<T> {

    final Subscriber<? super T> actual;
    Deque<T> buffer = new ConcurrentLinkedDeque<>();

    public ContinueWhileSubscriber(Subscriber<? super T> actual) {
        this.actual = actual;
    }

    @Override
    public void onNext(T t) {
        buffer.add(t);
        if (continuePredicate.call(t)) {
            while(!buffer.isEmpty())
                actual.onNext(buffer.poll());
        }
    }

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

    @Override
    public void onCompleted() {
        while (!buffer.isEmpty())
            actual.onNext(buffer.poll());
        buffer=null;
        actual.onCompleted();
    }
}
}




public static void main(String[] args) {
    BehaviorSubject<Integer> subject = BehaviorSubject.create();
    subject.asObservable()
            .doOnNext(v -> System.out.print("next "))
            .lift(ContinueWhile.create(i -> i%3==0))
            .subscribe(v -> System.out.print(v + ", "));
    for (int i = 0; i < 10; i++) {
        subject.onNext(i);
    }
}
}

Thanks to AndroidEx.