CODEWITHSUNDEEP
rx-javarx-java2
Oleksii Kolotylenko
Oleksii Kolotylenko

Reputation: 1090

How to skip first item if got second on some time period on RxJava

I need to perform some action in X seconds after received an event (represented any data item rxJava). But if got another event before X seconds ends, timer reset and action need to be performed again in X seconds.

Another words, we have UI elements with which user may interact. After X seconds of user's inactivity with this UI element we need to perform some action. Trivial solution could look like:

Disposable disp;
void init() {
             final Flowable<Boolean> flowable = 
mapDragOrZoomSubject.delay(TIME_X, TimeUnit.SECONDS)
                    .toFlowable(BackpressureStrategy.DROP);
        disp = flowable. ...subscribeWith(ACTION);  // <-- action in X seconds of inactivity
    }

void gotNewUserInteraction() {
    disp.dispose();
    init();
}

How to do this without disposing and re-creation Flowable basing on some rxJava operators?

Upvotes: 0

Views: 998

Answers (2)

TrogDor
TrogDor

Reputation: 1029

Try the throttleWithTimeout(...) operator:

@Test
public void testThrottleWithTimeout()
{
    Subject<String> eventSubject = PublishSubject.create();
    TestScheduler scheduler = new TestScheduler();

    TestObserver<String> observer = eventSubject
            .throttleWithTimeout( 1, TimeUnit.SECONDS, scheduler )
            .test();

    eventSubject.onNext( "foo" );
    scheduler.advanceTimeBy( 500, TimeUnit.MILLISECONDS );
    observer.assertNoValues();

    eventSubject.onNext( "bar" );
    scheduler.advanceTimeBy( 500, TimeUnit.MILLISECONDS );
    observer.assertNoValues();

    scheduler.advanceTimeBy( 500, TimeUnit.MILLISECONDS );
    observer.assertValue( "bar" );
}

Upvotes: 0

Garuno
Garuno

Reputation: 2200

I am not too familiar with rxJava, but the operator you are looking for is probably debounce. So maybe something like this:

Disposable disp;
void init() {
             final Flowable<Boolean> flowable = 
mapDragOrZoomSubject.delay(TIME_X, TimeUnit.SECONDS)
                    .toFlowable(BackpressureStrategy.DROP);
        disp = flowable
                 ...
                 .debounce(TIME_X, TimeUnit.SECONDS)
                 .subscribeWith(ACTION);  // <-- action in X seconds of inactivity
    }

void gotNewUserInteraction() {
    disp.dispose();
    init();
}

Upvotes: 1

Related Questions