RxSwift How to subscribe a observable only one time?

Question

I want to subscribe to an observable, but in some logic, I will re-subscribe to it. If I don't want to write some special logic for it, how can I dispose the last subscription when I add a new one? Or, when I subscribe to it, how can I know if this observable has been subscribed to already?

Maleki

Answer 1

The simplest solution for what you are looking for is indeed the method that they provided for just that - func take(_ count: Int).

Here is a playground sample:

import RxSwift

var variable = Variable(1)
variable.asObservable().take(1).subscribe(
    onNext: { print("onNext: \($0)") },
    onError: { print("onError: \($0)") },
    onCompleted: { print("onCompleted") },
    onDisposed: { print("onDisposed") })
variable.value = 2

print("done")

The results are:

onNext: 1
onCompleted
onDisposed
done

And yes, this is useful in places where you want to filter an event through a stream, in a location where it is awkward to store a subscription.

Aleta

Answer 2

Why do you want to do that? What problem are you specifically trying to solve?

The usual way to dispose subscriptions is to use a dispose bag.

func subscribeMyObservable()
{
    disposeBag = DisposeBag()

    myObservable
       .subscribe(onNext: { print("Whatever") }
       .addDisposableTo(disposeBag)
}

Notice how I recreate the disposable every time before subscribing? That would get rid of the previous subscriptions.

Subdual

Thanks for you answer .There is some logic that I need subscribe dynamic ally,so some time maybe resubscribe a observable, so I just think maybe there is some special observable ,if there is new subscribe ,it will auto-cancel the previous one .or some other smart way ,for now ,my solution is kind of like yours . – Maleki

You could take a look at the takeWhile or takeUntil operators, but I think the best way to do this is as I described. – Subdual

Answer 3

.switchLatest() operator is your friend

let eventObservableWrapper = PublishSubject<Observable<MyEvent>>()
let eventObservable = eventObservableWrapper.switchLatest() // use this one for subscriptions

// to switch to another observable, call -
eventObservableWrapper.onNext(someNewEventObservable)

see more about how switch works here - http://reactivex.io/RxJava/javadoc/rx/Observable.html#switchOnNext(rx.Observable)

Lutyens

Answer 4

What do you want to do if I´m not wrong is to subscribe and not unsubscribe once the items are emitted.

If that´s what you want you can use relay which you never gets unsubscribe.

  /**
     * ReplayRelay it works just like hot observables, once that an observer subscribe, the Relay will replay all items already emitted
     * to another observer.
     * it should return 1,2,3,4,5 for both observers.
     */
    @Test
    public void testReplayRelay() {
        ReplayRelay<String> relay = ReplayRelay.create();
        relay.subscribe(result -> System.out.println("Observer1:" + result));
        relay.call("1");
        relay.call("2");
        relay.call("3");
        relay.subscribe(result -> System.out.println("Observer2:" + result));
        relay.call("4");
        relay.call("5");
    }

You can see more exmaples here https://github.com/politrons/reactive/blob/master/src/test/java/rx/relay/Relay.java

Alatea

Answer 5

I did something similar to what @Odrakir proposed. I came across the same problem and still haven't found a better solution than this.

let disposeBag = DisposeBag()
var publishSubject = PublishSubject<Int>()
var count = 0

func setRx(toggle: Bool) {
    let tempSubject = PublishSubject<Int>()
    if toggle {
        tempSubject.subscribe(onNext: { n in
        self.count = n
        }).disposed(by: disposeBag)
    } else {
        tempSubject.dispose()
    }
    publishSubject = tempSubject
}

Buyers