Observable transformer that will act on event of the Lifecycle to auto dispose its subscriptions when the event occurs. By default ON_DESTROY

LifecycleObservableTransformer.java

package com.example.playground;

import android.arch.lifecycle.Lifecycle;
import android.arch.lifecycle.LifecycleObserver;
import android.arch.lifecycle.OnLifecycleEvent;
import android.support.annotation.NonNull;
import android.util.Log;

import java.lang.ref.WeakReference;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.disposables.CompositeDisposable;

/**
 * Transformer that will track observable's subscriptions and dispose them on given Event of the lifecycleRef object.
 * <p>
 * Copyright 2017 Mateusz Perlak - http://www.apache.org/licenses/LICENSE-2.0
 */
public class LifecycleObservableTransformer<T> implements ObservableTransformer<T, T>, LifecycleObserver {
    private static final String TAG = LifecycleObservableTransformer.class.getSimpleName();
    /**
     * Observed lifecycle reference
     */
    final WeakReference<Lifecycle> lifecycleRef;
    /**
     * CompositeDisposable for tracking the subscriptions.
     */
    final Map<Lifecycle.Event, CompositeDisposable> eventLifecycleDisposables = new ConcurrentHashMap<>();

    /**
     * Creates instance of lifecycleRef observer for the given Lifecycle and by default ON_DESTROY event.
     *
     * @param lifecycle
     */
    public LifecycleObservableTransformer(@NonNull Lifecycle lifecycle) {
        if (lifecycle == null) {
            throw new NullPointerException("lifecycleRef");
        }
        this.lifecycleRef = new WeakReference<>(lifecycle);
    }

    @OnLifecycleEvent(Lifecycle.Event.ON_DESTROY)
    public void onEventDestroy() {
        checkAndDispose(Lifecycle.Event.ON_DESTROY);
        // remove observer
    }

    @OnLifecycleEvent(Lifecycle.Event.ON_PAUSE)
    public void onEventPause() {
        checkAndDispose(Lifecycle.Event.ON_PAUSE);
    }

    @OnLifecycleEvent(Lifecycle.Event.ON_STOP)
    public void onEventStop() {
        checkAndDispose(Lifecycle.Event.ON_STOP);
    }

    /**
     * Checks if the event matching the provided event for disposal.
     *
     * @param occurredEvent
     */
    protected void checkAndDispose(Lifecycle.Event occurredEvent) {
        CompositeDisposable lifecycleDisposable = eventLifecycleDisposables.get(occurredEvent);
        if (lifecycleDisposable != null) {
            lifecycleDisposable.clear();
            // remove composite disposable if last one
            if (lifecycleDisposable.size()==0) {
                eventLifecycleDisposables.remove(occurredEvent);
            }
        }
        // remove last observer
        if (eventLifecycleDisposables.size()==0) {
            lifecycleRef.get().removeObserver(this);
        }
    }

    /**
     * Attaches subscription disposable on Observable's subscribe, for future disposal when event happens.
     *
     * @param upstream
     * @return
     */
    @Override
    public ObservableSource<T> apply(Observable<T> upstream) {
        return upstream.doOnSubscribe(disposable -> {
            Lifecycle lifecycle = lifecycleRef.get();
            if (lifecycle == null) {
                return;
            }
            Lifecycle.Event observedEvent = upDownEvent(lifecycle.getCurrentState());
            // on every subscription at current state dispose on down event
            if (observedEvent != null) {
                CompositeDisposable compositeDisposable = eventLifecycleDisposables.get(observedEvent);
                if (compositeDisposable == null) {
                    if (eventLifecycleDisposables.size()==0) {
                        lifecycle.addObserver(this);// first  so we subscribe
                    }
                    compositeDisposable = new CompositeDisposable();
                    eventLifecycleDisposables.put(observedEvent, compositeDisposable);
                }
                compositeDisposable.add(disposable);
            } else {
                Log.w(TAG, String.format("There is no down event for lifecycleRef object from current state: %s", lifecycle.getCurrentState()));
            }
        });
    }

    /**
     * Copied from LifecycleRegistry
     *
     * @param state
     * @return
     */
    public static Lifecycle.Event upDownEvent(Lifecycle.State state) {
        switch (state.ordinal()) {
            case 1: //Lifecycle.State.INITIALIZED
                return Lifecycle.Event.ON_DESTROY;
            case 2: //Lifecycle.State.CREATED
                return Lifecycle.Event.ON_STOP;
            case 3: //Lifecycle.State.STARTED
                return Lifecycle.Event.ON_PAUSE;
            default:
                return null;
        }
    }

}

usage-sample.java

        Observable.interval(1, TimeUnit.SECONDS)
                .compose(LifecycleObservableTransformer(this.getLifecycle())) // where this = LifecycleActivity or other Lifecycle aware component
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe { 
                    // do something 
                }