Comment faire pour arrêter et reprendre Observables.intervalle en prononçant les tiques

Cela va émettre un cocher toutes les 5 secondes.

Observable.interval(5, TimeUnit.SECONDS, Schedulers.io())
            .subscribe(tick -> Log.d(TAG, "tick = "+tick));

Pour l'arrêter, vous pouvez utiliser

Schedulers.shutdown();

Mais alors tous les Planificateurs s'arrête et il n'est pas possible de reprendre le tic-tac plus tard. Comment puis-je arrêter et reprendre le prononçant "gracieusement"`?

Je pense que la meilleure réponse à ce qui devrait être celui-ci: stackoverflow.com/questions/35782767/..., où Scan() est utilisé, donc il n'y a pas besoin d'accumuler une valeur à l'extérieur (comme suggéré dans la meilleure réponse)

InformationsquelleAutor Jan Seevers | 2016-02-15

36

Voici une solution possible:
```
class TickHandler {

    private AtomicLong lastTick = new AtomicLong(0L);
    private Subscription subscription;

    void resume() {
        System.out.println("resumed");
        subscription = Observable.interval(5, TimeUnit.SECONDS, Schedulers.io())
                                 .map(tick -> lastTick.getAndIncrement())
                                 .subscribe(tick -> System.out.println("tick = " + tick));
    }

    void stop() {
        if (subscription != null && !subscription.isUnsubscribed()) {
            System.out.println("stopped");
            subscription.unsubscribe();
        }
    }
}
```
- Si j'ai bien compris votre question, Observable.interval est un froid observables, de sorte que vous désabonner, il arrête ses émissions. Dans le même temps, vous pourriez probablement compter sur la mise en œuvre de Subscription (ou Disposable dans RxJava 2) pour y jeter la référence à la observables sur unsubscribe() (dispose()). Ou vous pouvez null le subscription de référence et de le rendre GC-admissibles pour assurer, de concert avec les observables sont stockées.
- La curiosité est question ici. Est-il important de déclarer lastTick comme Atomique ? Je pensais que toutes les volatile Atomiques et les choses ont été gérées par Rx directement.
- En général, lorsque plusieurs threads écrire à l'une des valeurs de champ qui dépend de la valeur précédente dans le domaine, les garanties prévues par la volatile ne sont pas assez fort, et l'accès doit être synchronisé ou organisés par le Java disponible simultanéité de l'Api. Plus sur le volatile tutorials.jenkov.com/java-concurrency/....
InformationsquelleAutor AndroidEx

Il y a quelques temps, j'ai également été à la recherche pour le type de RX "timer" des solutions, mais non d'entre eux a répondu à mes attentes. Donc là, vous pouvez trouver mon propre solution:

AtomicLong elapsedTime = new AtomicLong();
AtomicBoolean resumed = new AtomicBoolean();
AtomicBoolean stopped = new AtomicBoolean();

public Flowable<Long> startTimer() { //Create and starts timper
    resumed.set(true);
    stopped.set(false);
    return Flowable.interval(1, TimeUnit.SECONDS)
            .takeWhile(tick -> !stopped.get())
            .filter(tick -> resumed.get())
            .map(tick -> elapsedTime.addAndGet(1000));
}

public void pauseTimer() {
    resumed.set(false);
}

public void resumeTimer() {
    resumed.set(true);
}

public void stopTimer() {
    stopped.set(true);
}

public void addToTimer(int seconds) {
    elapsedTime.addAndGet(seconds * 1000);
}

Je ne peux pas croire que personne n'a upticked cette réponse. La accepté de répondre à crée un nouveau Observable après chaque pause et de ne pas accueillir la finale de champs dans les lambdas (Vous devez utiliser des champs de la classe). C'est une façon de trouver la meilleure solution.
C'est la seule réponse qui est bien pour l'utiliser dans le code de production.

InformationsquelleAutor Artur Szymański

val switch = new java.util.concurrent.atomic.AtomicBoolean(true)
val tick = new java.util.concurrent.atomic.AtomicLong(0L)

val suspendableObservable = 
  Observable.
    interval(5 seconds).
    takeWhile(_ => switch.get()).
    repeat.
    map(_ => tick.incrementAndGet())

Vous pouvez définir switch à false de suspendre le tic-tac et true la reprendre.

cette répétition est la clé. Merci

InformationsquelleAutor Sheng

Voici une autre façon de le faire, je pense.

Lorsque vous vérifiez le code source, vous trouverez intervalle() l'aide de la classe OnSubscribeTimerPeriodically. Le code de la clé ci-dessous.

@Override
public void call(final Subscriber<? super Long> child) {
    final Worker worker = scheduler.createWorker();
    child.add(worker);
    worker.schedulePeriodically(new Action0() {
        long counter;
        @Override
        public void call() {
            try {
                child.onNext(counter++);
            } catch (Throwable e) {
                try {
                    worker.unsubscribe();
                } finally {
                    Exceptions.throwOrReport(e, child);
                }
            }
        }

    }, initialDelay, period, unit);
}

Ainsi, vous pourrez voir, si tu veux cannel la boucle, qu'en jetant une nouvelle exception en onNext(). Exemple de code ci-dessous.

Observable.interval(1000, TimeUnit.MILLISECONDS)
            .subscribe(new Action1<Long>() {
                @Override
                public void call(Long aLong) {
                    Log.i("abc", "onNext");
                    if (aLong == 5) throw new NullPointerException();
                }
            }, new Action1<Throwable>() {
                @Override
                public void call(Throwable throwable) {
                    Log.i("abc", "onError");
                }
            }, new Action0() {
                @Override
                public void call() {
                    Log.i("abc", "onCompleted");
                }
            });

Puis vous verrez ceci:

08-08 11:10:46.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:47.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:48.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:49.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:50.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:51.008 28146-28181/net.bingyan.test I/abc: onNext
08-08 11:10:51.018 28146-28181/net.bingyan.test I/abc: onError

Les Exceptions doivent être utilisées uniquement pour le traitement des événements exceptionnels à ne pas mettre en œuvre une logique! Il crée des objets inutiles, il est mathématiquement lourde, et syntaxiquement à confusion

InformationsquelleAutor Chris Wong

Désolé c'est en RxJS au lieu de RxJava, mais le concept sera le même. J'ai adapté cette de apprenez-rxjs.io et ici c'est sur codepen.

L'idée est que vous commencez avec deux flux d'événements click, startClick$ et stopClick$. Cliquez sur chaque produit sur le stopClick$ stream obtenir mappé à un vide observables, et clique sur startClick$ chaque être substitués à la interval$ flux. Les deux en conséquence, les flux d'obtenir merge-d ensemble dans un observables-de-observables. En d'autres termes, un nouveau observable de l'un des deux types sera émis à partir de merge chaque fois qu'il y a un clic. La résultante observable passera par switchMap, qui commence à l'écoute de cette nouvelle observables et arrête d'écouter tout ce qu'il était à l'écoute de avant de. Switchmap permettra également de commencer à fusionner les valeurs de cette nouvelle observables sur ses flux de données existant.

Après le passage, scan seulement ne voit jamais l'incrémentation de la valeur émise par interval$, et il ne voit pas de valeurs lorsque "arrêt" a été cliqué.

Et jusqu'au premier clic, startWith va commencer émettant des valeurs de $interval, juste pour faire bouger les choses:

const start = 0;
const increment = 1;
const delay = 1000;
const stopButton = document.getElementById('stop');
const startButton = document.getElementById('start');
const startClick$ = Rx.Observable.fromEvent(startButton, 'click');
const stopClick$ = Rx.Observable.fromEvent(stopButton, 'click');
const interval$ = Rx.Observable.interval(delay).mapTo(increment);
const setCounter = newValue => document.getElementById("counter").innerHTML = newValue;
setCounter(start);

const timer$ = Rx.Observable

    //a "stop" click will emit an empty observable,
    //and a "start" click will emit the interval$ observable.  
    //These two streams are merged into one observable.
    .merge(stopClick$.mapTo(Rx.Observable.empty()), 
           startClick$.mapTo(interval$))

    //until the first click occurs, merge will emit nothing, so 
    //use the interval$ to start the counter in the meantime
    .startWith(interval$)

    //whenever a new observable starts, stop listening to the previous
    //one and start emitting values from the new one
    .switchMap(val => val)

    //add the increment emitted by the interval$ stream to the accumulator
    .scan((acc, curr) => curr + acc, start)

    //start the observable and send results to the DIV
    .subscribe((x) => setCounter(x));

Et voici le code HTML

<html>
<body>
  <div id="counter"></div>
  <button id="start">
    Start
  </button>
  <button id="stop">
    Stop
  </button>
</body>
</html>

InformationsquelleAutor mikebridge

Vous pouvez utiliser takeWhile et boucle jusqu'à ce que les conditions est vrai

Observable.interval(1, TimeUnit.SECONDS)
        .takeWhile {
            Log.i(TAG, " time " + it)
            it != 30L
        }
        .subscribe(object : Observer<Long> {
            override fun onComplete() {
                Log.i(TAG, "onComplete " + format.format(System.currentTimeMillis()))
            }

            override fun onSubscribe(d: Disposable) {
                Log.i(TAG, "onSubscribe " + format.format(System.currentTimeMillis()))
            }

            override fun onNext(t: Long) {
                Log.i(TAG, "onNext " + format.format(System.currentTimeMillis()))
            }

            override fun onError(e: Throwable) {
                Log.i(TAG, "onError")
                e.printStackTrace()
            }

        });

InformationsquelleAutor Alier

Vous devez vous connecter pour publier un commentaire.