Is it necessary to unsubscribe

Question

I am having hard time understanding RX. In the following case, is it necessary to unsubscribe? Is there a way to automatically unsubscribe after the "call" function was executed?

Observable.create(new Observable.OnSubscribe<NumberInfo>() {
        @Override
        public void call(Subscriber<? super NumberInfo> subscriber) {
            try {
                // Store data to db
            } catch (Exception e) {
                Log.e(TAG, "Downloaded numberInfo was not added to cache.", e);
            }
        }
    }).subscribeOn(Schedulers.newThread())
    .subscribe(); 

I don't want to observe for any result that's why I omitted the classical .observeOn(AndroidSchedulers.mainThread())

Thx for explanation.

Answer 1

According to Rx contract, when the Observable fires onCompleted, the Observer unsubscribes. In your case, the contract is not respected because there is no subscriber.onCompleted() in your code.

If you just need something like "Fire and forget", you could try just:

Schedulers.io().createWorker().schedule(new Action0() {
    @Override
    public void call() {
        try {
            // Store data to db
        } catch (Exception e) {
            Log.e(TAG, "Downloaded numberInfo was not added to cache.", e);
        }
    }
});

It will execute on I/O Scheduler and your UI thread is safe.

IMO you should always have a return value. Your Store data to db routing surely has some return value, like a long specifying the row number or a boolean that indicates success. Having this approach, you can create a proper method:

public Observable<Long> storeToDb(final SomethingToStore storeMe) {
    return Observable
            .create(new Observable.OnSubscribe<Long>() {
                @Override
                public void call(Subscriber<? super Long> subscriber) {
                    long row = syncStore(storeMe);

                    if (row == -1) {
                        subscriber.onError(new Throwable("Cannot store " + storeMe.toString + " to DB."));
                    }

                    subscriber.onNext(row);
                    subscriber.onCompleted();
                }
            }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

And you could use it like this:

storeToDb(storeThis)
        .subscribe(new Observer<Long>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {
                Log.e("STORING", "Something went south: " + e.getMessage());
            }

            @Override
            public void onNext(Long row) {
                Log.d("STORING", "Everything has been stored as record number: " + row);
            }
        });