java类rx.subscriptions.Subscriptions的实例源码

/**
 * This methods observes a firebase query and returns back ONCE
 * an Observable of the {@link DataSnapshot}
 * when the firebase client uses a {@link ValueEventListener}
 *
 * @param firebaseRef {@link Query} this is reference of a Firebase Query
 * @return an {@link rx.Observable} of datasnapshot to use
 */
public Observable<DataSnapshot> observeSingleValue(final Query firebaseRef) {
  return Observable.create(new Observable.OnSubscribe<DataSnapshot>() {
    @Override public void call(final Subscriber<? super DataSnapshot> subscriber) {
      final ValueEventListener listener = new ValueEventListener() {
        @Override public void onDataChange(DataSnapshot dataSnapshot) {
          subscriber.onNext(dataSnapshot);
          subscriber.onCompleted();
        }

        @Override public void onCancelled(DatabaseError error) {
          FirebaseDatabaseErrorFactory.buildError(subscriber, error);
        }
      };

      firebaseRef.addListenerForSingleValueEvent(listener);

      // When the subscription is cancelled, remove the listener
      subscriber.add(Subscriptions.create(new Action0() {
        @Override public void call() {
          firebaseRef.removeEventListener(listener);
        }
      }));
    }
  }).compose(this.<DataSnapshot>applyScheduler());
}

/**
 * Observes the authentication state for {@link com.google.firebase.auth.FirebaseAuth}
 *
 * @return {@link rx.Observable} of {@link com.google.firebase.auth.FirebaseUser}
 */
public Observable<FirebaseUser> observeAuthState() {
  return Observable.create(new Observable.OnSubscribe<FirebaseUser>() {
    @Override public void call(final Subscriber<? super FirebaseUser> subscriber) {
      final AuthStateListener authStateListener = new AuthStateListener() {
        @Override public void onAuthStateChanged(@NonNull FirebaseAuth firebaseAuth) {
          FirebaseUser firebaseUser = firebaseAuth.getCurrentUser();
          if (firebaseUser == null) {
            FirebaseAuthErrorFactory.createSignOutError(subscriber);
          } else {
            subscriber.onNext(firebaseUser);
          }
        }
      };
      firebaseAuth.addAuthStateListener(authStateListener);
      subscriber.add(Subscriptions.create(new Action0() {
        @Override public void call() {
          firebaseAuth.removeAuthStateListener(authStateListener);
        }
      }));
    }
  });
}

@Override
public void call(final Subscriber<? super Intent> subscriber) {
    final BroadcastReceiver broadCastReceiver=new BroadcastReceiver() {
        @Override
        public void onReceive(Context context, Intent intent) {
            subscriber.onNext(intent);
        }
    };
    final Subscription subscription= Subscriptions.create(new Action0() {
        @Override
        public void call() {
            context.unregisterReceiver(broadCastReceiver);

        }
    });
    subscriber.add(subscription);
    context.registerReceiver(broadCastReceiver, intentFilter, permission, schedulerHandler);
}

private Subscription unsubscribeInUiThread(final Action0 unsubscribe) {
    return Subscriptions.create(new Action0() {

        @Override public void call() {
            if (Looper.getMainLooper() == Looper.myLooper()) {
                unsubscribe.call();
            } else {
                final Scheduler.Worker inner = AndroidSchedulers.mainThread().createWorker();
                inner.schedule(new Action0() {
                    @Override public void call() {
                        unsubscribe.call();
                        inner.unsubscribe();
                    }
                });
            }
        }
    });
}

@Inject
public PreferencesService(@ApplicationContext Context context) {
    mPrefs = context.getSharedPreferences(PREFS_FILE_NAME, Context.MODE_PRIVATE);

    mChangedKeys = Observable.create(new Observable.OnSubscribe<String>() {
        @Override
        public void call(final Subscriber<? super String> subscriber) {
            final OnSharedPreferenceChangeListener listener = new OnSharedPreferenceChangeListener() {
                @Override
                public void onSharedPreferenceChanged(SharedPreferences sharedPreferences, String key) {
                    subscriber.onNext(key);
                }
            };

            Subscription subscription = Subscriptions.create(new Action0() {
                @Override public void call() {
                    mPrefs.unregisterOnSharedPreferenceChangeListener(listener);
                }
            });
            subscriber.add(subscription);

            mPrefs.registerOnSharedPreferenceChangeListener(listener);
        }
    }).share();
}

/**
 * Show an AlertDialog with the {@code title} and the {@code message}. The Alert dialog has an
 * "yes" button and a "no" button.
 *
 * @param title Title to apply on AlertDialog
 * @param message Message to asSnack on AlertDialog
 *
 * @return A Observable that shows the dialog when subscribed and return the action made by
 * user. This action is represented by EResponse
 *
 * @see EResponse
 */
public static Observable<EResponse> createGenericYesNoCancelMessage(@NonNull Context context,
    @Nullable String title, @Nullable String message) {
  return Observable.create((Subscriber<? super EResponse> subscriber) -> {
    final AlertDialog dialog = new AlertDialog.Builder(context).setTitle(title)
        .setMessage(message)
        .setPositiveButton(android.R.string.yes, (listener, which) -> {
          subscriber.onNext(EResponse.YES);
          subscriber.onCompleted();
        })
        .setNegativeButton(android.R.string.no, (listener, which) -> {
          subscriber.onNext(EResponse.NO);
          subscriber.onCompleted();
        })
        .setOnCancelListener(listener -> {
          subscriber.onNext(EResponse.CANCEL);
          subscriber.onCompleted();
        })
        .create();
    // cleaning up
    subscriber.add(Subscriptions.create(() -> dialog.dismiss()));
    dialog.show();
  })
      .subscribeOn(AndroidSchedulers.mainThread());
}

/**
 * Show an AlertDialog with the {@code title} and the {@code message}. The Alert dialog has an
 * "ok" button.
 *
 * @param title Title to apply on AlertDialog
 * @param message Message to asSnack on AlertDialog
 *
 * @return A Observable that shows the dialog when subscribed and return the action made by
 * user. This action is represented by EResponse
 *
 * @see EResponse
 */
public static Observable<EResponse> createGenericOkCancelMessage(Context context, String title,
    String message) {
  return Observable.create((Subscriber<? super EResponse> subscriber) -> {
    final AlertDialog dialog = new AlertDialog.Builder(context).setTitle(title)
        .setMessage(message)
        .setPositiveButton(android.R.string.ok, (listener, which) -> {
          subscriber.onNext(EResponse.YES);
          subscriber.onCompleted();
        })
        .setNegativeButton(android.R.string.cancel, (dialogInterface, i) -> {
          subscriber.onNext(EResponse.CANCEL);
          subscriber.onCompleted();
        })
        .create();
    // cleaning up
    subscriber.add(Subscriptions.create(() -> dialog.dismiss()));
    dialog.show();
  });
}

public static Observable<EResponse> createGenericContinueMessage(Context context,
    @Nullable View view, String title, String message, @StringRes int buttonText) {
  return Observable.create((Subscriber<? super EResponse> subscriber) -> {
    AlertDialog.Builder builder = new AlertDialog.Builder(context).setTitle(title)
        .setMessage(message)
        .setPositiveButton(buttonText, (dialogInterface, i) -> {
          subscriber.onNext(EResponse.YES);
          subscriber.onCompleted();
        });
    if (view != null) {
      builder.setView(view);
    }
    AlertDialog alertDialog = builder.create();
    subscriber.add(Subscriptions.create(() -> alertDialog.dismiss()));
    alertDialog.show();
  });
}

public static Observable<EResponse> createGenericContinueCancelMessage(Context context,
    String title, String message) {
  return Observable.create((Subscriber<? super EResponse> subscriber) -> {
    final AlertDialog ad = new AlertDialog.Builder(context).setTitle(title)
        .setMessage(message)
        .setPositiveButton(android.R.string.ok, (dialog, which) -> {
          subscriber.onNext(EResponse.YES);
          subscriber.onCompleted();
        })
        .setNegativeButton(android.R.string.cancel, (dialogInterface, i) -> {
          subscriber.onNext(EResponse.NO);
          subscriber.onCompleted();
        })
        .setOnCancelListener(dialog -> {
          subscriber.onNext(EResponse.CANCEL);
          subscriber.onCompleted();
        })
        .create();
    // cleaning up
    subscriber.add(Subscriptions.create(() -> ad.dismiss()));
    ad.show();
  });
}

public static Observable<EResponse> createGenericOkCancelMessage(Context context, String title,
    @StringRes int message, @StringRes int okMessage, @StringRes int cancelMessage) {
  return Observable.create((Subscriber<? super EResponse> subscriber) -> {
    final AlertDialog ad = new AlertDialog.Builder(context).setTitle(title)
        .setMessage(message)
        .setPositiveButton(okMessage, (dialog, which) -> {
          subscriber.onNext(EResponse.YES);
          subscriber.onCompleted();
        })
        .setNegativeButton(cancelMessage, (dialogInterface, i) -> {
          subscriber.onNext(EResponse.NO);
          subscriber.onCompleted();
        })
        .setOnCancelListener(dialog -> {
          subscriber.onNext(EResponse.CANCEL);
          subscriber.onCompleted();
        })
        .create();
    // cleaning up
    subscriber.add(Subscriptions.create(() -> ad.dismiss()));
    ad.show();
  });
}