Rx Valeurs d'émission observables périodiquement

L'une des options est d'utiliser Observables.d'intervalle et de la vérification de l'état de l'utilisateur lorsque les intervalles sont émis:

     Observable<Long> interval = Observable.interval(1, TimeUnit.SECONDS);

    //pulling the user data
    Observable<Observable<String>> userObservable = interval.map(new Func1<Long, Observable<String>>() {
        Random random = new Random();
        @Override
        public Observable<String> call(Long tick) {
            //here you are pulling user data; you should do it asynchronously - rx way - because the interval is using Schedulers.computation which is not best suited for doing io operations
            switch(random.nextInt(10)){
                case 0://suppose this is for cases when network in  not available or exception happens
                    return Observable.<String>just(null);
                case 1:
                case 2:
                    return Observable.just("Alice");
                default:
                    return Observable.just("Bob");
            }
        }
    });

    Observable<String> flatUsers = userObservable.flatMap(new Func1<Observable<String>, Observable<? extends String>>() {
        @Override
        public Observable<? extends String> call(Observable<String> stringObservable) {
            return stringObservable;
        }
    });

    //filter valid data
    Observable<String> usersWithoutErrors = flatUsers.filter(new Func1<String, Boolean>() {
        @Override
        public Boolean call(String s) {
            return s != null;
        }
    });

    //publish only changes
    Observable<String> uniqueUsers = usersWithoutErrors.distinctUntilChanged();

Vous pouvez faire encore plus simple si votre networkStatusObservable émet des événements au moins aussi souvent que vous avez besoin de vérifier les données de l'utilisateur

 networkStatusObservable.sample(1,TimeUnit.Seconds).filter(/*the best is to filter only connected state */).map(/*now start pulling the user data*/)

Enfin, vous pouvez peut créer observables qui utilise le planificateur d'émettre l'utilisateur périodiquement les états - reportez-vous à Les planificateurs de la documentation pour apprendre l'algorithme adapter à vos besoins le meilleur:

public abstract class ScheduledOnSubscribe<T> implements Observable.OnSubscribe<T>{
private final Scheduler scheduler;
private final long initialDelay;
private final long period;
private final TimeUnit unit;
public ScheduledOnSubscribe(Scheduler scheduler, long initialDelay, long period, TimeUnit unit) {
this.scheduler = scheduler;
this.initialDelay = initialDelay;
this.period = period;
this.unit = unit;
}
abstract T next() throws Exception;
@Override
public void call(final Subscriber<? super T> subscriber) {
final Scheduler.Worker worker = scheduler.createWorker();
subscriber.add(worker);
worker.schedulePeriodically(new Action0() {
@Override
public void call() {
try {
subscriber.onNext(next());
} catch (Throwable e) {
try {
subscriber.onError(e);
} finally {
worker.unsubscribe();
}
}
}
}, initialDelay, period, unit);
}
}
//And here is the sample usage
Observable<String> usersObservable = Observable.create(new ScheduledOnSubscribe(Schedulers.io(), 1, 1, TimeUnit.SECONDS ){
Random random = new Random();
@Override
String next() throws Exception {
//if you use Schedulers.io, you can call the remote service synchronously
switch(random.nextInt(10)){
case 0:
return null;
case 1:
case 2:
return "Alice";
default:
return "Bob";
}
}
});

D'accord, je vais poster ma propre solution, peut-être quelqu'un qui va en tirer profit. Je vais seulement après la partie traitant de la question, en omettant le HTTP et la mise en cache des trucs. Voici comment je le fais:

private ConnectableObservable<Long> createNetworkBoundHeartbeatObservable(final Observable<Boolean> networkStatusObservable,
final Observable<Boolean> pauseResumeObservable) {
final Observable<Boolean> pausableHeartbeatObservable = Observable.combineLatest(networkStatusObservable, pauseResumeObservable,
new Func2<Boolean, Boolean, Boolean>() {
@Override
public Boolean call(Boolean networkAvailable, Boolean mustPause) {
return mustPause && networkAvailable;
}
}
).distinctUntilChanged();
final Observable<Boolean> hasToResumeObservable = pausableHeartbeatObservable.filter(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean networkAvailable) {
return networkAvailable;
}
});
final Observable<Boolean> hasToStopObservable = pausableHeartbeatObservable.filter(new Func1<Boolean, Boolean>() {
@Override
public Boolean call(Boolean networkAvailable) {
return !networkAvailable;
}
});
return pausableHeartbeatObservable.concatMap(new Func1<Boolean, Observable<Long>>() {
@Override
public Observable<Long> call(Boolean shouldResumeRequests) {
if (shouldResumeRequests) {
long timeToUpdate;
final Date oldestModifiedExpiresAt = cache.oldestModifiedExpiresAt();
timeToUpdate = Math.max(0, oldestModifiedExpiresAt.getTime() - System.currentTimeMillis());
Log.d(TAG, String.format("Have to restart updates, %d seconds till next update", timeToUpdate / SECOND_IN_MILLIS));
return Observable
.timer(timeToUpdate, SECONDS_TO_EXPIRE * SECOND_IN_MILLIS, TimeUnit.MILLISECONDS)
.takeUntil(hasToStopObservable);
} else {
Log.d(TAG, "Have to pause updates");
return Observable.<Long>never().takeUntil(hasToResumeObservable);
}
}
}).multicast(PublishSubject.<Long>create());
}

Comme vous pouvez le voir, les conditions pour mettre en pause ou de reprendre les mises à jour de devenir un peu plus compliqué, avec une nouvelle Observables ajoutés à l'appui de la pause lors de l'application va à fond.

Ensuite à la base de la solution est la concatMap opération qui émet de la Observables de manière séquentielle (d'où concatMap, pas flatMap, voir à cette question: Quelle est la différence entre concatMap et flatMap dans RxJava). Il émet soit interval ou never Observablesselon que les mises à jour devraient être poursuivis ou mis en pause. Ensuite, chaque émis Observable est takenUntil 'un en face de' Observable émet de la nouvelle valeur.

ConnectableObservable est retourné parce que l'créée Observable est chaud, et que tous les abonnés de souscrire à elle avant qu'elle ne commence à émettre quelque chose, sinon événements initiaux pourraient être perdues. J'appelle connect sur elle plus tard.

Je vais accepter ma ou une autre réponse basées sur les votes, le cas échéant.