Java 类io.reactivex.SingleEmitter 实例源码

/**
 * The returned Single emits one Query result as a List.
 */
public static <T> Single<List<T>> single(final Query<T> query) {
    return Single.create(new SingleOnSubscribe<List<T>>() {
        @Override
        public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
            final DataSubscription dataSubscription = query.subscribe().single().observer(new DataObserver<List<T>>() {
                @Override
                public void onData(List<T> data) {
                    if (!emitter.isDisposed()) {
                        emitter.onSuccess(data);
                    }
                }
            });
            emitter.setCancellable(new Cancellable() {
                @Override
                public void cancel() throws Exception {
                    dataSubscription.cancel();
                }
            });
        }
    });
}

private void single() {
    Single.create(new SingleOnSubscribe<String>() {
        @Override
        public void subscribe(SingleEmitter<String> e) throws Exception {
            e.onSuccess("1");
            e.onSuccess("2");
            e.onSuccess("3");
            e.onSuccess("4");
        }
    }).subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            Log.e(TAG, "accept: success " + s);
        }
    }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {

        }
    });

}

@SuppressLint("NewApi")
public Single<Boolean> isAlive() {
    if (RXSDebug.isDebug()) Timber.tag(TAG).v("isAlive()");
    return Single
            .create((SingleEmitter<Boolean> emitter) -> {
                if (ApiWrap.hasOreo()) {
                    emitter.onSuccess(process.isAlive());
                } else {
                    try {
                        process.exitValue();
                        emitter.onSuccess(false);
                    } catch (IllegalThreadStateException e) {
                        emitter.onSuccess(true);
                    }
                }
            })
            .subscribeOn(Schedulers.io());
}

static void onResolutionResult(String observableId, int resultCode) {
    if (observableMap.containsKey(observableId)) {
        SettingsCheckHandleSingleOnSubscribe observable = observableMap.get(observableId).get();

        if (observable != null && observable.emitterWeakRef != null) {
            SingleEmitter<Boolean> observer = observable.emitterWeakRef.get();

            if (observer != null) {
                observer.onSuccess(resultCode == Activity.RESULT_OK);
            }
        }

        observableMap.remove(observableId);
    }

    observableMapCleanup();
}

@Override
public final void subscribe(SingleEmitter<T> emitter) throws Exception {
    final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter));

    try {
        apiClient.connect();
    } catch (Throwable ex) {
        emitter.onError(ex);
    }

    emitter.setCancellable(new Cancellable() {
        @Override
        public void cancel() throws Exception {
            if (apiClient.isConnected()) {
                RxLocationSingleOnSubscribe.this.onUnsubscribed(apiClient);
            }

            apiClient.disconnect();
        }
    });
}

public void saveSnapshotToStorage(final Bitmap bitmap, final String title) {
    final int quality = SPUtils.getBoolean(Constants.SP_HD_SCREENSHOT) ? 100 : 90;

    mComposite.add(Single.create(new SingleOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {
            String address = IOUtils.saveBitmapToExternalStorage(bitmap, title, quality);
            e.onSuccess(address);
        }
    })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                mBaseView.onSnapshotSavedSucceeded(s);
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                mBaseView.onSnapshotSavedFailed(throwable);
            }
        }));
}

public Single<List<CList>> getList(final int mFilterType, final String orderBy, final String limit, final String skip) {

        return Single.create(new SingleOnSubscribe<List<CList>>() {
            @Override
            public void subscribe(SingleEmitter<List<CList>> emitter) throws Exception {
                Cursor fetchCursor = getCursorByType(mFilterType, orderBy);

                if (fetchCursor == null) {
                    emitter.onError(new Exception("Cursor is null"));
                    return;
                }


                Log.d(TAG, "|Start Filter Type  = " + mFilterType + " " + new Date(System.currentTimeMillis()).toString() + "\n Cursor Count" + fetchCursor.getCount());
                emitter.onSuccess(new ArrayList<>(fillMap(fetchCursor, mFilterType).values()));

            }
        });
    }

/************************************************************/
public Single<Void> deleteCollBookInRx(CollBookBean bean) {
    return Single.create(new SingleOnSubscribe<Void>() {
        @Override
        public void subscribe(SingleEmitter<Void> e) throws Exception {
            //查看文本中是否存在删除的数据
            deleteBook(bean.get_id());
            //删除任务
            deleteDownloadTask(bean.get_id());
            //删除目录
            deleteBookChapter(bean.get_id());
            //删除CollBook
            mCollBookDao.delete(bean);
            e.onSuccess(new Void());
        }
    });
}

@Override
public void subscribe(final SingleEmitter<QuerySnapshot> emitter) throws Exception {
    OnCompleteListener<QuerySnapshot> listener = new OnCompleteListener<QuerySnapshot>() {
        @Override
        public void onComplete(@NonNull Task<QuerySnapshot> task) {
            if (!emitter.isDisposed()) {
                if (task.isSuccessful()) {
                    emitter.onSuccess(task.getResult());

                } else {
                    emitter.onError(task.getException());

                }
            }

        }
    };

    collectionReference.get()
            .addOnCompleteListener(listener);
}

public Single<T> toSingle() {
    return Single.create(new SingleOnSubscribe<T>() {
        @Override
        public void subscribe(final @NonNull SingleEmitter<T> e) throws Exception {
            enqueue(new Callback<T>() {
                        @Override
                        public void onResponse(@NonNull T data) {
                            e.onSuccess(data);
                        }
                    }
                    , new ErrorCallback() {
                        @Override
                        public void onError(@NonNull Throwable error) {
                            e.onError(error);
                        }
                    }
            );
        }
    });
}

static void onResolutionResult(String observableId, int resultCode) {
    if (observableMap.containsKey(observableId)) {
        SettingsCheckHandleSingleOnSubscribe observable = observableMap.get(observableId).get();

        if (observable != null && observable.emitterWeakRef != null) {
            SingleEmitter<Boolean> observer = observable.emitterWeakRef.get();

            if (observer != null) {
                observer.onSuccess(resultCode == Activity.RESULT_OK);
            }
        }

        observableMap.remove(observableId);
    }

    observableMapCleanup();
}

@Override
public final void subscribe(SingleEmitter<T> emitter) throws Exception {
    final GoogleApiClient apiClient = createApiClient(new ApiClientConnectionCallbacks(emitter));

    try {
        apiClient.connect();
    } catch (Throwable ex) {
        emitter.onError(ex);
    }

    emitter.setCancellable(() -> {
        if (apiClient.isConnected()) {
            onUnsubscribed(apiClient);
        }

        apiClient.disconnect();
    });
}

/**
 * @param emit
 * @param <R>
 * @return
 */
@NonNull
@CheckReturnValue
public static <R> OnCompleteListener<R> listener(@NonNull final SingleEmitter<R> emit) {
    return new OnCompleteListener<R>() {
        @Override
        public void onComplete(@NonNull final Task<R> task) {
            if (!emit.isDisposed()) {
                if (task.isSuccessful()) {
                    emit.onSuccess(task.getResult());
                } else {
                    Exception e = task.getException();
                    emit.onError(e != null ? e : new RuntimeException());
                }
            }
        }
    };
}

@Override public void subscribe(final SingleEmitter<String> emitter) {
  OnCompleteListener<GetTokenResult> listener = new OnCompleteListener<GetTokenResult>() {
    @Override public void onComplete(@NonNull Task<GetTokenResult> task) {
      if (!task.isSuccessful()) {
        if (!emitter.isDisposed()) {
          emitter.onError(task.getException());
        }
        return;
      }

      if (!emitter.isDisposed()) {
        emitter.onSuccess(task.getResult().getToken());
      }
    }
  };

  user.getToken(forceRefresh).addOnCompleteListener(listener);
}

@Override public void subscribe(final SingleEmitter<FirebaseUser> emitter) {
  final OnCompleteListener<AuthResult> listener = new OnCompleteListener<AuthResult>() {
    @Override public void onComplete(@NonNull Task<AuthResult> task) {
      if (!task.isSuccessful()) {
        if (!emitter.isDisposed()) {
          emitter.onError(task.getException());
        }
        return;
      }

      if (!emitter.isDisposed()) {
        emitter.onSuccess(task.getResult().getUser());
      }
    }
  };
  createAuthTask(instance).addOnCompleteListener(listener);
}

public static <T> OnCompleteListener<T> listener(@NonNull final SingleEmitter<T> emitter) {
  return new OnCompleteListener<T>() {
    @Override public void onComplete(@NonNull Task<T> task) {
      if (!task.isSuccessful()) {
        if (!emitter.isDisposed()) {
          emitter.onError(task.getException());
        }
        return;
      }

      if (!emitter.isDisposed()) {
        emitter.onSuccess(task.getResult());
      }
    }
  };
}

@Override public void subscribe(final SingleEmitter<DataSnapshot> emitter) {
  final ValueEventListener listener = new ValueEventListener() {
    @Override public void onDataChange(DataSnapshot dataSnapshot) {
      if (!emitter.isDisposed()) {
        emitter.onSuccess(dataSnapshot);
      }
    }

    @Override public void onCancelled(DatabaseError databaseError) {
      if (!emitter.isDisposed()) {
        emitter.onError(databaseError.toException());
      }
    }
  };

  ref.addListenerForSingleValueEvent(listener);
}

public static <T> Single<T> single(final OnSubscribeAction<T> subscribe) {
    final RxActionDelegate<T> delegate = new RxActionDelegate<>();
    return Single.<T>create(new SingleOnSubscribe<T>() {
        @Override
        public void subscribe(@NonNull final SingleEmitter<T> emitter) throws Exception {
            delegate.setDelegate(new ActionDelegate<T>() {
                @Override
                public void onSuccess(T result) {
                    emitter.onSuccess(result);
                }

                @Override
                public void onError(Exception e) {
                    emitter.onError(e);
                }
            });
            subscribe.subscribe(delegate);
        }
    }).doOnDispose(new Action() {
        @Override
        public void run() throws Exception {
            delegate.cancel();
        }
    });
}

public static <T> Single<Result<T>> singleWrapped(final OnSubscribeAction<T> subscribe) {
    final RxActionDelegate<T> delegate = new RxActionDelegate<>();
    return Single.<Result<T>>create(new SingleOnSubscribe<Result<T>>() {
        @Override
        public void subscribe(@NonNull final SingleEmitter<Result<T>> emitter) throws Exception {
            delegate.setDelegate(new ActionDelegate<T>() {
                @Override
                public void onSuccess(T result) {
                    emitter.onSuccess(new Result<T>(result));
                }

                @Override
                public void onError(Exception e) {
                    emitter.onSuccess(new Result<T>(e));
                }
            });
            subscribe.subscribe(delegate);
        }
    }).doOnDispose(new Action() {
        @Override
        public void run() throws Exception {
            delegate.cancel();
        }
    });
}

/**
 * Given an operation that takes a {@link MaybeConsumer<T>}, create a JavaRX
 * {@link Single<T>} that produces the value passed to the MaybeConsumer.
 *
 * Example:
 * <pre>
 *     // log the name of the experiment with a given id
 *     DataController dc = getDataController();
 *     MaybeConsumers.buildSingle(mc -> dc.getExperimentById(id, mc))
 *                   .subscribe(experiment -> log("Name: " + experiment.getName()));
 * </pre>
 */
public static <T> Single<T> buildSingle(io.reactivex.functions.Consumer<MaybeConsumer<T>> c) {
    return Single.create(new SingleOnSubscribe<T>() {
        @Override
        public void subscribe(SingleEmitter<T> emitter) throws Exception {
            c.accept(new MaybeConsumer<T>() {
                @Override
                public void success(T value) {
                    emitter.onSuccess(value);
                }

                @Override
                public void fail(Exception e) {
                    emitter.onError(e);
                }
            });
        }
    });
}

@Test public void verifyCancellation() {
  final AtomicInteger i = new AtomicInteger();
  //noinspection unchecked because Java
  Single<Integer> source = Single.create(new SingleOnSubscribe<Integer>() {
    @Override public void subscribe(SingleEmitter<Integer> e) {
      e.setCancellable(new Cancellable() {
        @Override public void cancel() {
          i.incrementAndGet();
        }
      });
    }
  });
  MaybeSubject<Integer> lifecycle = MaybeSubject.create();
  source.as(AutoDispose.<Integer>autoDisposable(lifecycle))
      .subscribe();

  assertThat(i.get()).isEqualTo(0);
  assertThat(lifecycle.hasObservers()).isTrue();

  lifecycle.onSuccess(0);

  // Verify cancellation was called
  assertThat(i.get()).isEqualTo(1);
  assertThat(lifecycle.hasObservers()).isFalse();
}

private Single<Ignore> connectService() {
  return Single.create(new SingleOnSubscribe<Ignore>() {
    @Override public void subscribe(final SingleEmitter<Ignore> emitter) throws Exception {
      if (serviceConnection == null) {
        serviceConnection = new ServiceConnection() {
          @Override public void onServiceDisconnected(ComponentName name) {
            rxBillingServiceLogger.log(getTargetClassName(), "Service disconnected");
            appBillingService = null;
            emitter.onError(
                new RxBillingServiceException(RxBillingServiceError.SERVICE_DISCONNECTED));
          }

          @Override public void onServiceConnected(ComponentName name, final IBinder service) {
            rxBillingServiceLogger.log(getTargetClassName(), "Service connected");
            appBillingService = IInAppBillingService.Stub.asInterface(service);
            emitter.onSuccess(Ignore.Get);
          }
        };

        bindService();
      } else {
        emitter.onSuccess(Ignore.Get);
      }
    }
  });
}

/**
 * @param query
 * @return
 */
@NonNull
@CheckReturnValue
public static Single<DataSnapshot> single(@NonNull final Query query) {
    return Single.create(new SingleOnSubscribe<DataSnapshot>() {
        @Override
        public void subscribe(
                @NonNull final SingleEmitter<DataSnapshot> emit) throws Exception {
            final ValueEventListener listener = listener(emit);

            emit.setCancellable(new Cancellable() {
                @Override
                public void cancel() throws Exception {
                    query.removeEventListener(listener);
                }
            });

            query.addListenerForSingleValueEvent(listener);
        }
    });
}

/**
 * @param emit
 * @return
 */
@NonNull
@CheckReturnValue
public static ValueEventListener listener(@NonNull final SingleEmitter<DataSnapshot> emit) {
    return new ValueEventListener() {
        @Override
        public void onDataChange(DataSnapshot dataSnapshot) {
            if (!emit.isDisposed()) {
                emit.onSuccess(dataSnapshot);
            }
        }

        @Override
        public void onCancelled(DatabaseError e) {
            if (!emit.isDisposed()) {
                emit.onError(e.toException());
            }
        }
    };
}

@Override @NonNull public Single<List<T>> get() {
  return Single.create(new SingleOnSubscribe<List<T>>() {
    @Override public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
      runInReadLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (!file.exists()) {
            emitter.onSuccess(Collections.<T>emptyList());
            return;
          }

          List<T> list = converter.read(file, type);
          if (list == null) list = Collections.emptyList();
          emitter.onSuccess(list);
        }
      });
    }
  });
}

@Override @NonNull public Single<List<T>> observePut(@NonNull final List<T> list) {
  assertNotNull(list, "list");

  return Single.create(new SingleOnSubscribe<List<T>>() {
    @Override public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (!file.exists() && !file.createNewFile()) {
            throw new IOException("Could not create file for store.");
          }

          converterWrite(list, converter, type, file);
          emitter.onSuccess(list);
          updateSubject.onNext(list);
        }
      });
    }
  });
}

@Override @NonNull public Single<List<T>> observeClear() {
  return Single.create(new SingleOnSubscribe<List<T>>() {
    @Override public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (file.exists() && !file.delete()) {
            throw new IOException("Clear operation on store failed.");
          }

          emitter.onSuccess(Collections.<T>emptyList());
          updateSubject.onNext(Collections.<T>emptyList());
        }
      });
    }
  });
}

@Override @NonNull public Single<List<T>> observeAdd(@NonNull final T value) {
  assertNotNull(value, "value");

  return Single.create(new SingleOnSubscribe<List<T>>() {
    @Override public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (!file.exists() && !file.createNewFile()) {
            throw new IOException("Could not create file for store.");
          }

          List<T> originalList = converter.read(file, type);
          if (originalList == null) originalList = Collections.emptyList();

          List<T> result = new ArrayList<T>(originalList.size() + 1);
          result.addAll(originalList);
          result.add(value);

          converterWrite(result, converter, type, file);
          emitter.onSuccess(result);
          updateSubject.onNext(result);
        }
      });
    }
  });
}

@Override @NonNull public Single<List<T>> observeRemove(final int position) {
  return Single.create(new SingleOnSubscribe<List<T>>() {
    @Override public void subscribe(final SingleEmitter<List<T>> emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          List<T> originalList = converter.read(file, type);
          if (originalList == null) originalList = Collections.emptyList();

          List<T> modifiedList = new ArrayList<T>(originalList);
          modifiedList.remove(position);

          converterWrite(modifiedList, converter, type, file);
          emitter.onSuccess(modifiedList);
          updateSubject.onNext(modifiedList);
        }
      });
    }
  });
}

@Override @NonNull public Single<T> observePut(@NonNull final T value) {
  assertNotNull(value, "value");

  return Single.create(new SingleOnSubscribe<T>() {
    @Override public void subscribe(final SingleEmitter<T> emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (!file.exists() && !file.createNewFile()) {
            throw new IOException("Could not create file for store.");
          }

          converterWrite(value, converter, type, file);
          emitter.onSuccess(value);
          updateSubject.onNext(new ValueUpdate<T>(value));
        }
      });
    }
  });
}

@NonNull private String createDirectory(SingleEmitter<String> e) {
  File appDir =
      new File(Environment.getExternalStorageDirectory() + "/AppPermissionsExtractedApk");
  if (!appDir.exists()) {
    if (!appDir.mkdir()) {
      e.onError(new Exception("Error creating directory"));
    }
  }
  return appDir.getPath();
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, SingleEmitter<Status> emitter) {
    //noinspection MissingPermission
    setupLocationPendingResult(
            LocationServices.FusedLocationApi.requestLocationUpdates(apiClient, locationRequest, pendingIntent),
            SingleResultCallBack.get(emitter)
    );
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, SingleEmitter<LocationSettingsResult> emitter) {
    setupLocationPendingResult(
            LocationServices.SettingsApi.checkLocationSettings(apiClient, locationSettingsRequest),
            SingleResultCallBack.get(emitter)
    );
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, SingleEmitter<Status> emitter) {
    setupLocationPendingResult(
            LocationServices.FusedLocationApi.removeLocationUpdates(apiClient, pendingIntent),
            SingleResultCallBack.get(emitter)
    );
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, SingleEmitter<Status> emitter) {
    setupLocationPendingResult(
            LocationServices.FusedLocationApi.flushLocations(apiClient),
            SingleResultCallBack.get(emitter)
    );
}

@Override
protected void onGoogleApiClientReady(GoogleApiClient apiClient, SingleEmitter<Boolean> emitter) {
    //noinspection MissingPermission
    LocationAvailability locationAvailability = LocationServices.FusedLocationApi.getLocationAvailability(apiClient);

    if (locationAvailability != null) {
        emitter.onSuccess(locationAvailability.isLocationAvailable());
    } else {
        emitter.onSuccess(false);
    }
}

public static Single<Submission> getSubmission(final RedditClient redditClient) {
    return Single.create(new SingleOnSubscribe<Submission>() {
        @Override
        public void subscribe(SingleEmitter<Submission> e) throws Exception {
            try {
                e.onSuccess(redditClient.getRandomSubmission());
            } catch (Exception ex) {
                e.onError(ex);
            }
        }
    });
}

protected @OnClick(R.id.tv_save_image)
void saveImage() {
    BitmapPool pool = BitmapPool.getPool();
    final String fileName = MD5.generate(mUrl);
    BitmapWrapper bitmapWrapper = pool.get(fileName, true, true);

    final Bitmap bitmap = bitmapWrapper.getBitmap();

    mDisposable = Single.create(new SingleOnSubscribe<String>() {
        @Override
        public void subscribe(@NonNull SingleEmitter<String> e) throws Exception {
            String address = IOUtils.saveBitmapToExternalStorage(bitmap, fileName, 100);
            e.onSuccess(address);
        }
    })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                ToastUtils.showShortToast(getString(R.string.image_saved) + IOUtils.PIC_DIR);
                MediaScannerConnection.scanFile(Aequorea.getApp()
                    .getApplicationContext(), new String[]{Environment.getExternalStorageDirectory()
                    .getPath().concat(IOUtils.PIC_DIR)}, null, new MediaScannerConnection.OnScanCompletedListener() {
                    @Override
                    public void onScanCompleted(final String path, Uri uri) {
                        // do nothing.
                    }
                });

                dismiss();
            }
        }, new Consumer<Throwable>() {
            @Override
            public void accept(Throwable throwable) throws Exception {
                ToastUtils.showShortToast(getString(R.string.save_failed) + throwable.getMessage());
                dismiss();
            }
        });
}

@Override public Single<Boolean> checkInternetConnectivity(final String host, final int port,
                                                           final int timeoutInMs, final ErrorHandler errorHandler) {
    checkGeneralPreconditions(host, port, timeoutInMs, errorHandler);

    return Single.create(new SingleOnSubscribe<Boolean>() {
        @Override public void subscribe(@NonNull SingleEmitter<Boolean> emitter) throws Exception {
            emitter.onSuccess(isConnected(host, port, timeoutInMs, errorHandler));
        }
    });
}

@Override public Single<Boolean> checkInternetConnectivity(final String host, final int port,
                                                           final int timeoutInMs, final ErrorHandler errorHandler) {
    checkGeneralPreconditions(host, port, timeoutInMs, errorHandler);

    return Single.create(new SingleOnSubscribe<Boolean>() {
        @Override public void subscribe(@NonNull SingleEmitter<Boolean> emitter) throws Exception {
            emitter.onSuccess(isConnected(host, port, timeoutInMs, errorHandler));
        }
    });
}