RxJava: Aufruf onError ohne Abschluss / Abmeldung

So Spiele ich schon mit diesem für einige Zeit, und ich glaube nicht, dass es möglich ist, in der Weise Sie es tun. Aufruf onError oder onCompleted beenden Sie den stream, spiegeln die done Flagge innerhalb der SafeSubscriber wrapper, und es ist nicht nur ein Weg, um es zurückzusetzen.

Ich sehe 2 Optionen zur Verfügung - weder finde ich besonders elegant, aber funktioniert.

1 - UnsafeSubscribe. Vielleicht nicht die beste Idee, aber es funktioniert, weil statt der Verpackung Ihrer Subscriber im SafeSubscriber, ruft es direkt. Am besten Lesen Sie die Javadoc, um zu sehen, wenn das OK ist für Sie. Oder, wenn Sie abenteuerlustig fühlen, schreiben Sie Ihre eigenen SafeSubscriber wo setzen Sie die done-flag oder ähnliches. Mit deinem Beispiel so nennen:

observe.unsafeSubscribe(...)

2 - Umsetzung etwas ähnliches wie dieses Beispiel. Ich Schätze es ist in C#, aber es sollte lesbar sein. Einfach ausgedrückt - Sie wollen ein Pair<T, Exception> Klasse, und dann zu fordern, anstatt onError rufen onNext und legen Sie die Ausnahme Seite Ihres pair. Ihre Abonnenten haben, um ein wenig klüger zu werden, um zu überprüfen, für jede Seite des Paares, und Sie müssen einige Daten-transformation zwischen Ihrer Quelle Observable und die Observable<Pair<T, Exception>>, aber ich kann nicht sehen, warum es nicht funktionieren wird.

Ich wäre wirklich daran interessiert, sehen einen anderen Weg, dies zu tun, wenn jemand keine.

Hoffe, das hilft,

Wird

Als @Wird darauf hingewiesen, dass man nicht direkt nennen onError ohne die Beendigung des beobachtbaren. Da kann man nur rufen onNext haben, beschloss ich, ein Benachrichtigung zu wickeln Sie den Wert und die throwable-in einem einzigen Objekt.

import rx.*;
import rx.functions.Action0;
import rx.schedulers.Schedulers;
import rx.subscriptions.Subscriptions;

import java.util.concurrent.TimeUnit;

public class PollingObservable {
    public static <T> Observable<Notification<T>> create(Observable<T> observable, long delayTime, TimeUnit unit) {
        return Observable.create(new OnSubscribePolling<>(observable, delayTime, unit));
    }

    private static class OnSubscribePolling<T> implements Observable.OnSubscribe<Notification<T>> {
        private Subscription subscription;
        private Subscription innerSubscription;
        private Scheduler.Worker worker = Schedulers.newThread().createWorker();

        private Observable<T> observable;
        private long delayTime;
        private TimeUnit unit;

        private boolean isUnsubscribed = false;

        public OnSubscribePolling(final Observable<T> observable, long delayTime, TimeUnit unit) {
            this.observable = observable;
            this.delayTime = delayTime;
            this.unit = unit;
        }

        @Override
        public void call(final Subscriber<? super Notification<T>> subscriber) {
            subscription = worker.schedule(new Action0() {
                @Override
                public void call() {
                    schedule(subscriber, true);
                }
            });

            subscriber.onStart();
            subscriber.add(Subscriptions.create(new Action0() {
                @Override
                public void call() {
                    isUnsubscribed = true;

                    subscription.unsubscribe();
                    if (innerSubscription != null) {
                        innerSubscription.unsubscribe();
                    }
                }
            }));
        }

        private void schedule(final Subscriber<? super Notification<T>> subscriber, boolean immediately) {
            if (isUnsubscribed) {
                return;
            }

            long delayTime = immediately ? 0 : this.delayTime;
            subscription = worker.schedule(createInnerAction(subscriber), delayTime, unit);
        }

        private Action0 createInnerAction(final Subscriber<? super Notification<T>> subscriber) {
            return new Action0() {
                @Override
                public void call() {
                    innerSubscription = observable.subscribe(new Observer<T>() {
                        @Override
                        public void onCompleted() {
                            schedule(subscriber, false);
                        }

                        @Override
                        public void onError(Throwable e) {
                            subscriber.onNext(Notification.<T>createOnError(e));
                            schedule(subscriber, false);
                        }

                        @Override
                        public void onNext(T t) {
                            subscriber.onNext(Notification.createOnNext(t));
                        }
                    });
                }
            };
        }
    }
}

Diesen zu nutzen, können Sie entweder die Benachrichtigung direkt:

PollingObservable.create(service.getStatus(), 5, TimeUnit.SECONDS)
    .subscribe(new Action1<Notification<Status>>() {
        @Override
        public void call(Notification<Status> notification) {
            switch (notification.getKind()) {
                case OnNext:
                    Status status = notification.getValue();
                    //handle onNext event
                    break;
                case OnError:
                    Throwable throwable = notification.getThrowable();
                    //handle onError event
                    break;
            }
        }
    });

Oder verwenden Sie den akzeptieren - Methode für die Benachrichtigung auf einen regulären Beobachten:

PollingObservable.create(service.getStatus(), 5, TimeUnit.SECONDS)
        .subscribe(new Action1<Notification<Status>>() {
            @Override
            public void call(Notification<Status> notification) {
               notification.accept(statusObserver);
            }
        });

Observer<Status> statusObserver = new Observer<Status>() {
    //...
}

UPDATE 2015-02-24

Scheint es, dass die polling-beobachtbare hatte nicht richtig funktioniert, manchmal, weil die innere beobachtbaren nennen würde onComplete oder onError auch nach es gewesen war abgemeldet, also der Umschuldung selbst. Ich fügte hinzu, die isUnsubscribed Flagge zu verhindern.