Managing Subscriptions in Android with RxJava

ActivitySubscriptionManager.java

/*
 * Copyright 2014 Prateek Srivastava
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.f2prateek.couchpotato.data.rx;

import android.app.Activity;

public class ActivitySubscriptionManager extends SubscriptionManager<Activity> {
  public ActivitySubscriptionManager(Activity instance) {
    super(instance);
  }

  @Override protected boolean validate(final Activity activity) {
    return !activity.isFinishing();
  }
}

BaseFragment.java

public class BaseFragment extends Fragment {
  private final SubscriptionManager<Fragment> = new FragmentSubscriptionManger(this);
  
  ...
  @Override public void onPause() {
    subscriptionManager.unsubscribeAll();
    super.onPause();
  }
}

FragmentSubscriptionManager.java

/*
 * Copyright 2014 Prateek Srivastava
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.f2prateek.couchpotato.data.rx;

import android.app.Fragment;

public class FragmentSubscriptionManager extends SubscriptionManager<Fragment> {
  public FragmentSubscriptionManager(Fragment instance) {
    super(instance);
  }

  @Override protected boolean validate(final Fragment fragment) {
    return fragment.isAdded() && !fragment.getActivity().isFinishing();
  }
}

SubscriptionManager.java

/*
 * Copyright 2014 Prateek Srivastava
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.f2prateek.couchpotato.data.rx;

import rx.Observable;
import rx.Observer;
import rx.Subscription;
import rx.android.observables.Assertions;
import rx.functions.Func1;
import rx.operators.OperatorConditionalBinding;
import rx.subscriptions.CompositeSubscription;

import static rx.android.schedulers.AndroidSchedulers.mainThread;

/**
 * A class to help manage subscriptions. It will automatically unsubscribe any subscription if the
 * predicate does not validate.
 */
public abstract class SubscriptionManager<T> {
  private final T instance;
  private final Func1<T, Boolean> predicate = new Func1<T, Boolean>() {
    @Override
    public Boolean call(T object) {
      return validate(object);
    }
  };
  private final CompositeSubscription subscriptions = new CompositeSubscription();

  public SubscriptionManager(T instance) {
    this.instance = instance;
  }

  public <O> void subscribe(final Observable<O> source, final Observer<O> observer) {
    Assertions.assertUiThread();
    Subscription subscription = source.observeOn(mainThread())
        .lift(new OperatorConditionalBinding<O, T>(instance, predicate))
        .subscribe(observer);
    subscriptions.add(subscription);
  }

  public void unsubscribe() {
    subscriptions.unsubscribe();
  }

  /**
   * Return a {@link rx.functions.Func1} implementation that will be a predicate for {@link
   * OperatorConditionalBinding}. If the predicate fails to validate, the sequence unsubscribes
   * itself and releases the bound reference.
   */
  protected abstract boolean validate(final T object);
}

I can not find OperatorConditionalBinding