关于在Chocolat或Sublime Text 2中使用Objective-C的任何一般性评论也将受到欢迎.

和我的options,这是一个开始

…This method always returns an object.
The data in the persistent store
represented by objectID is assumed to
exist—if it does not,the returned
object throws an exception when you
access any property (that is,when the
fault is fired). The benefit of this
behavior is that it allows you to
create and use faults,then create the
underlying rows later or in a separate
context.

在Apple的’Core Recipes’示例应用程序中,该方法的结果用于填充NSFetchRequest,然后使用请求的结果,并对此结果进行注释：

// first get the object into the context
 Recipe *recipeFault = (Recipe *)[context objectWithID:objectID];

 // this only creates a fault,which may NOT resolve to an object (for example,if the ID is for
 // an objec that has been deleted already):  create a fetch request to get the object for real
 NSFetchRequest *request = [[[NSFetchRequest alloc] init] autorelease];
 [request setEntity: [NSEntityDescription entityForName:@"Recipe" inManagedobjectContext:context]];
 nspredicate *predicate = [nspredicate predicateWithFormat: @"(self == %@)",recipeFault];
                    [request setPredicate: predicate];

我看过很多例子(其他代码和apple的’iClass’),其中objectWithID的结果直接使用 – 意思是,它的属性被访问并快速处理.

objectWithID是否应始终被视为“可能存在”的对象？

我问,因为我刚刚遇到这个并没有反对它的存在.

您可以像对待它一样对待它,访问其属性等等,因为在后台Core Data将访问数据存储来为您的请求提供服务.但是,如果商店中不存在该对象,您将获得异常.

Here’s Apple’s documentation explaining faults(这是objectWithID：返回给你).

@Test
public void shouldContinueRunningAfterExceptionInTask(){
    AtomicBoolean firstRun = new AtomicBoolean(true);
    CompletableSubject completedSecondRun = CompletableSubject.create();
    KeyedTaskBuffer<String,Integer> taskBuffer = new KeyedTaskBuffer<>(BUFFER_SIZE_2,MAX_WAIT_MILLIS_100,(key,batch) -> {
                if(firstRun.getAndSet(false)){
                    throw new RuntimeException("Test exception");
                }else{
                    completedSecondRun.onComplete();
                }
            });
    taskBuffer.addData(KEY_A,1);
    taskBuffer.addData(KEY_B,1);

    //unit test can only finish if this is completed
    completedSecondRun.timeout(1000,TimeUnit.MILLISECONDS).blockingAwait();
}

@Test public void test() {


        CompletableSubject subject = CompletableSubject.create();
        Completable source = subject.hide();


        TestObserver testObserver = new TestObserver();
        Compositedisposable composite = new Compositedisposable();
        disposable disposable = source
                .compose(disposableAttach.<String>to(composite))
                .subscribeWith(testObserver);

        assertTrue(composite.size() == 1);
        composite.dispose();
        assertTrue(composite.size() == 0);
        assertTrue(composite.isdisposed());
        assertTrue(disposable.isdisposed());
        assertTrue(testObserver.isdisposed());
    }

/**
 * Apply all the queued operations to the database.
 *
 * @return Observable,either immediately available (when offline) or triggered on
 * Firebase success/failure event.
 */
public Completable write() {
    CompletableSubject subject = CompletableSubject.create();

    LOGGER.info("Writing with online={},operations in queue: {}",ServerConnection.isOnline(),OPERATIONS_IN_FLIGHT);
    OPERATIONS_IN_FLIGHT.incrementAndGet();

    Throwable writerStackTrace = new Throwable();

    mRoot.updateChildren(mData)
            .addOnSuccessListener((final Void p) -> subject.onComplete())
            .addOnFailureListener((final Exception e) -> {
                e.setStackTrace(writerStackTrace.getStackTrace());
                LOGGER.error("Failed to save {}",mData,e);
                subject.onError(e);
            })
            .addOnCompleteListener(t -> OPERATIONS_IN_FLIGHT.decrementAndGet());

    if (ServerConnection.isOnline()) {
        return subject;
    } else {
        return Completable.complete();
    }
}

@Test public void autodispose_withMaybe_normal() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  MaybeSubject<Integer> lifecycle = MaybeSubject.create();
  source.as(autodisposable(lifecycle))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  // Got the event
  source.onComplete();
  o.assertOnComplete();

  // nothing more,lifecycle disposed too
  o.assertNoMoreEvents();
  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();
}

@Test public void autodispose_withMaybe_interrupted() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  MaybeSubject<Integer> lifecycle = MaybeSubject.create();
  source.as(autodisposable(lifecycle))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  // Lifecycle ends
  lifecycle.onSuccess(2);
  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();

  // Event if upstream emits,no one is listening
  source.onComplete();
  o.assertNoMoreEvents();
}

@Test public void autodispose_withProvider_completion() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  MaybeSubject<Integer> scope = MaybeSubject.create();
  ScopeProvider provider = makeProvider(scope);
  source.as(autodisposable(provider))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(scope.hasObservers()).isTrue();

  source.onComplete();
  o.assertOnComplete();

  o.assertNoMoreEvents();
  assertthat(source.hasObservers()).isFalse();
  assertthat(scope.hasObservers()).isFalse();
}

@Test public void autodispose_withProvider_interrupted() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  MaybeSubject<Integer> scope = MaybeSubject.create();
  ScopeProvider provider = makeProvider(scope);
  source.as(autodisposable(provider))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(scope.hasObservers()).isTrue();

  scope.onSuccess(1);

  // All disposed
  assertthat(source.hasObservers()).isFalse();
  assertthat(scope.hasObservers()).isFalse();

  // No one is listening
  source.onComplete();
  o.assertNoMoreEvents();
}

@Test public void autodispose_withLifecycleProvider_completion() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  BehaviorSubject<Integer> lifecycle = BehaviorSubject.createDefault(0);
  LifecycleScopeProvider<Integer> provider = makeLifecycleProvider(lifecycle);
  source.as(autodisposable(provider))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  lifecycle.onNext(1);

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  source.onComplete();
  o.assertOnComplete();

  o.assertNoMoreEvents();
  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();
}

@Test public void autodispose_withProviderAndNoOpPlugin_withoutStarting_shouldFailSilently() {
  AutodisposePlugins.setoutsideLifecycleHandler(new Consumer<OutsideLifecycleException>() {
    @Override public void accept(OutsideLifecycleException e) { }
  });
  BehaviorSubject<Integer> lifecycle = BehaviorSubject.create();
  LifecycleScopeProvider<Integer> provider = TestUtil.makeLifecycleProvider(lifecycle);
  CompletableSubject source = CompletableSubject.create();
  TestObserver<Void> o = source
          .as(autodisposable(provider))
          .test();

  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();
  o.assertNovalues();
  o.assertNoErrors();
}

@Test public void autodispose_withProviderAndNoOpPlugin_afterEnding_shouldFailSilently() {
  AutodisposePlugins.setoutsideLifecycleHandler(new Consumer<OutsideLifecycleException>() {
    @Override public void accept(OutsideLifecycleException e) {
      // Noop
    }
  });
  BehaviorSubject<Integer> lifecycle = BehaviorSubject.createDefault(0);
  lifecycle.onNext(1);
  lifecycle.onNext(2);
  lifecycle.onNext(3);
  LifecycleScopeProvider<Integer> provider = TestUtil.makeLifecycleProvider(lifecycle);
  CompletableSubject source = CompletableSubject.create();
  TestObserver<Void> o = source
          .as(autodisposable(provider))
          .test();

  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();
  o.assertNovalues();
  o.assertNoErrors();
}

@Test public void autodispose_withProviderAndplugin_withoutStarting_shouldFailWithWrappedExp() {
  AutodisposePlugins.setoutsideLifecycleHandler(new Consumer<OutsideLifecycleException>() {
    @Override public void accept(OutsideLifecycleException e) {
      // Wrap in an IllegalStateException so we can verify this is the exception we see on the
      // other side
      throw new IllegalStateException(e);
    }
  });
  BehaviorSubject<Integer> lifecycle = BehaviorSubject.create();
  LifecycleScopeProvider<Integer> provider = TestUtil.makeLifecycleProvider(lifecycle);
  TestObserver<Void> o = CompletableSubject.create()
            .as(autodisposable(provider))
            .test();

  o.assertNovalues();
  o.assertError(new Predicate<Throwable>() {
    @Override public boolean test(Throwable throwable) {
      return throwable instanceof IllegalStateException
          && throwable.getCause() instanceof OutsideLifecycleException;
    }
  });
}

public MessageAckNowledger(SqsQueue<T> sqsQueue,String receiptId,Instant expirationTime) {
    this.expirationTime = expirationTime;
    this.sqsQueue = sqsQueue;
    this.receiptId = receiptId;
    this.ackModeSingle = SingleSubject.create();
    this.ackingComplete = CompletableSubject.create();

    Duration duration = Duration.between(Instant.Now(),expirationTime);
    Completable.timer(duration.toMillis(),TimeUnit.MILLISECONDS).subscribe(this::ignore);
}

/**
 * Returns a Completable that terminates when the given CompletionStage terminates.
 * @param future the source CompletionStage instance
 * @return the new Completable instance
 */
public static Completable fromFuture(CompletionStage<?> future) {
    CompletableSubject cs = CompletableSubject.create();

    future.whenComplete((v,e) -> {
        if (e != null) {
            cs.onError(e);
        } else {
            cs.onComplete();
        }
    });

    return cs;
}

@Test public void autodispose_withLifecycleProvider_interrupted() {
  RecordingObserver<Integer> o = new RecordingObserver<>(LOGGER);
  CompletableSubject source = CompletableSubject.create();
  BehaviorSubject<Integer> lifecycle = BehaviorSubject.createDefault(0);
  LifecycleScopeProvider<Integer> provider = makeLifecycleProvider(lifecycle);
  source.as(autodisposable(provider))
      .subscribe(o);
  o.takeSubscribe();

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  lifecycle.onNext(1);

  assertthat(source.hasObservers()).isTrue();
  assertthat(lifecycle.hasObservers()).isTrue();

  lifecycle.onNext(3);

  // All disposed
  assertthat(source.hasObservers()).isFalse();
  assertthat(lifecycle.hasObservers()).isFalse();

  // No one is listening
  source.onComplete();
  o.assertNoMoreEvents();
}

@Test public void autodispose_withScopeProviderCompleted_shouldNotReportDoubleSubscriptions() {
  TestObserver<Void> o = CompletableSubject.create()
            .as(autodisposable(ScopeProvider.UNBOUND))
            .test();
  o.assertNovalues();
  o.assertNoErrors();

  rule.assertNoErrors();
}

@Test public void unbound_shouldStillPassValues() {
  TestObserver<Void> o = CompletableSubject.create()
          .as(autodisposable(ScopeProvider.UNBOUND))
          .test();

  o.onComplete();
  o.assertComplete();
}

@Default
public CompletableSubject getResultSubject() {
    return CompletableSubject.create();
}

@Default
public CompletableSubject getResultSubject() {
    return CompletableSubject.create();
}

@Override
public Completable deleteMessage(String receiptHandle) {
    return Completable.defer(() -> delegate.deleteMessage(receiptHandle))
            .retry(this::shouldRetry)
            .subscribeWith(CompletableSubject.create());//convert to Hot completable
}

@Override
public Completable changeMessageVisibility(String receiptHandle,Duration newVisibility) {
    return Completable.defer(() -> delegate.changeMessageVisibility(receiptHandle,newVisibility))
            .retry(this::shouldRetry)
            .subscribeWith(CompletableSubject.create());//convert to Hot completable
}

private Observable<String> showNameInputDialog(String prefix,MaterialDialog.InputCallback textWatcher) {
    final PublishSubject<String> input = PublishSubject.create();
    DialogUtils.showDialog(new ThemeColorMaterialDialogBuilder(mContext).title(R.string.text_name)
            .inputType(InputType.TYPE_CLASS_TEXT)
            .alwaysCallInputCallback()
            .input(getString(R.string.text_please_input_name),prefix,false,textWatcher)
            .onPositive(new MaterialDialog.SingleButtonCallback() {
                @Override
                public void onClick(@NonNull MaterialDialog dialog,@NonNull DialogAction which) {
                    input.onNext(dialog.getInputEditText().getText().toString());
                    input.onComplete();
                }
            })
            .build());
    return input;
}

/**
 * PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者。
 * 需要注意的是，PublishSubject可能会一创建完成就立刻开始发射数据（除非你可以阻止它发生），
 * 因此这里有一个风险：在Subject被创建后到有观察者订阅它之前这个时间段内，一个或多个数据可能会丢失。
 * 如果要确保来自原始Observable的所有数据都被分发:
 * 1.使用Create创建那个Observable以便手动给它引入"冷"Observable的行为（当所有观察者都已经订阅时才开始发射数据）
 * 2.改用ReplaySubject。
 *
 * 如果原始的Observable因为发生了一个错误而终止，PublishSubject将不会发射任何数据，只是简单的向前传递这个错误通知。
 */
private void doSomeWork() {

    PublishSubject<Integer> source = PublishSubject.create();

    source.subscribe(getFirstObserver()); // it will get 1,2,3,4 and onComplete

    source.onNext(1);
    source.onNext(2);
    source.onNext(3);

    /*
     * it will emit 4 and onComplete for second observer also.
     */
    source.subscribe(getSecondobserver());

    source.onNext(4);
    source.onComplete();
}

void onRequestPermissionsResult(String permissions[],int[] grantResults,boolean[] shouldShowRequestPermissionRationale) {
    for (int i = 0,size = permissions.length; i < size; i++) {
        Log.i("","onRequestPermissionsResult  " + permissions[i]);
        // Find the corresponding subject
        PublishSubject<Permission> subject = mSubjects.get(permissions[i]);
        if (subject == null) {
            // No subject found
            Log.e(RxPermissions.TAG,"RxPermissions.onRequestPermissionsResult invoked but didn't find the corresponding permission request.");
            return;
        }
        mSubjects.remove(permissions[i]);
        boolean granted = grantResults[i] == PackageManager.PERMISSION_GRANTED;
        subject.onNext(new Permission(permissions[i],granted,shouldShowRequestPermissionRationale[i]));
        subject.onComplete();
    }
}

private void monitor(InputStream input,PublishSubject<String> subject) {
    BufferedReader reader = new BufferedReader(new InputStreamReader(input,encoding));
    final int BUFFERSIZE = 4096;
    char[] buffer = new char[BUFFERSIZE];
    while (true) {
        try {
            if (Thread.interrupted()) {
                subject.onComplete();
                return;
            }
            int read = reader.read(buffer,BUFFERSIZE);
            if (read == -1) {
                subject.onComplete();
                return;
            }

            subject.onNext(new String(buffer,read));
        } catch (IOException e) {
            subject.onError(e);
            return;
        }
    }
}

@Override
public void downloadProducts(DownloadProductsRequest request,StreamObserver<Product> responSEObserver) {
  PublishSubject<Product> productPublishSubject = PublishSubject.create();
  productPublishSubject
      .doOnNext(product -> {
        responSEObserver.onNext(product);
        counter.labels("downloadProducts","success");
      })
      .doOnComplete(() -> responSEObserver.onCompleted())
      .doOnError(t -> {
        responSEObserver.onError(t);
        counter.labels("downloadProducts","Failed");
      })
      .subscribe();
  productDao.downloadProducts(request,productPublishSubject);
}

private void initView() {
        subject = PublishSubject.create();
        subject.debounce(0,TimeUnit.MILLISECONDS)
//                .filter(new Predicate<Float>() {
//                    @Override
//                    public boolean test(Float brightness) throws Exception {
//                        return true;
//                    }
//                })
                .distinctUntilChanged()
                .switchMap(new Function<Float,ObservableSource<ColorMatrixColorFilter>>() {
                    @Override
                    public ObservableSource<ColorMatrixColorFilter> apply(Float value) throws Exception {
                        return postBrightness(value);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<ColorMatrixColorFilter>() {
                    @Override
                    public void accept(ColorMatrixColorFilter colorMatrixColorFilter) throws Exception {
                        setColorFilter(colorMatrixColorFilter);
                    }
                });
    }

@Inject
TranscodingServiceImpl(ExternalProcessService externalProcessService,TranscoderSettings transcoderSettings,MediaScanSettings mediaScanSettings,OutputParser parser) {
    this.externalProcessService = externalProcessService;
    this.transcoderSettings = transcoderSettings;
    this.mediaScanSettings = mediaScanSettings;

    this.parser = parser;

    this.publisher = PublishSubject.create().toSerialized();

    publisher.ofType(TranscodeTask.class)
             .skipwhile(o -> isActive())
             .observeOn(Schedulers.computation())
             .subscribeOn(Schedulers.io())
             .subscribe(this::prepareTranscode);
}

@Override
public View onCreateView(LayoutInflater inflater,ViewGroup container,Bundle savedInstanceState) {
    // Inflate the layout for this fragment
    View view = inflater.inflate(R.layout.fragment_two_way_data_biding,container,false);
    unbinder = ButterKnife.bind(this,view);

    publishSubject = PublishSubject.create();

    publishSubject.subscribe(new Consumer<String>() {
        @Override
        public void accept(String s) throws Exception {
            mTvAddResult.setText(s);
        }
    });

    onNumChanged();

    MetaddLeft.requestFocus();

    return view;
}

@Test
public void onUserQueryEvent_emptyQueryString_shouldDonothingWithView() {
  // given
  PublishSubject<UserQueryEvent> userQuery$ = PublishSubject.create();
  UserListPresenter presenter = new UserListPresenter(
      userQuery$,userService,() -> mock(UserView.class),() -> mock(UserPresenter.class),DEFAULT_PAGE_SIZE,DEFAULT_USER_SEARCH_LIMIT
  );
  presenter.start(view);
  UserQueryEvent event = new UserQueryEvent(" "); // empty string

  // when
  userQuery$.onNext(event);

  // then
  verifyNoMoreInteractions(view,userService);
}

void onRequestPermissionsResult(String permissions[],size = permissions.length; i < size; i++) {
        log("onRequestPermissionsResult  " + permissions[i]);
        // Find the corresponding subject
        PublishSubject<Permission> subject = mSubjects.get(permissions[i]);
        if (subject == null) {
            // No subject found
            Log.e(RxPermissions.TAG,"RxPermissions.onRequestPermissionsResult invoked but didn't find the corresponding permission request.");
            return;
        }
        mSubjects.remove(permissions[i]);
        boolean granted = (grantResults[i] == PackageManager.PERMISSION_GRANTED) && (PermissionChecker.checkSelfPermission(getContext(),permissions[i]) == PermissionChecker.PERMISSION_GRANTED);
        subject.onNext(new Permission(permissions[i],shouldShowRequestPermissionRationale[i]));
        subject.onComplete();
    }
}

/**
 * 接受消息
 *
 * @param tag      标志
 * @param callBack 回调
 */
public <T> disposableObserver registerNoThread(@NonNull final Object tag,@NonNull final RxBusCallBack<T> callBack) {
    RxBusEvent rxBusEvent = rxBusEventArrayMap.get(tag);
    if (RxUtils.isEmpty(rxBusEvent)) {
        rxBusEvent = new RxBusEvent();
        rxBusEvent.subject = PublishSubject.create().toSerialized();
        rxBusEvent.disposable =
                rxBusEvent.subject
                        .ofType(callBack.busOfType())
                        .subscribeWith(new RxBusObserver<T>() {
                            @Override
                            public void onError(@io.reactivex.annotations.NonNull Throwable e) {
                                super.onError(e);
                                callBack.onBusError(e);
                            }

                            @Override
                            public void onNext(@io.reactivex.annotations.NonNull T t) {
                                super.onNext(t);
                            }
                        });
        rxBusEventArrayMap.put(tag,rxBusEvent);
    }
    return rxBusEvent.disposable;
}

@Test
public void onUserSelected_view_shouldPostUserSelectedEvent() {
  // given
  TestObserver<UserSelectedEvent> userSelected$ = TestObserver.create();
  User user = mock(User.class);
  given(user.getLogin()).willReturn("foo");
  PublishSubject<Trigger> userSelectionIntent = PublishSubject.create();
  given(view.userSelection$()).willReturn(userSelectionIntent);

  UserPresenter presenter = new UserPresenter(Sink.of(userSelected$));
  presenter.start(user,view);

  // when
  fire(userSelectionIntent);

  // then
  userSelected$.assertValueCount(1);
  UserSelectedEvent event = userSelected$.values().get(0);
  assertthat(event.getUser().getLogin()).isEqualTo("foo");
}

public Observable<Boolean> refreshOnlinestatus() {
    PublishSubject<Boolean> online = PublishSubject.create();
    mRetrofit.create(UserApi.class)
            .me()
            .subscribeOn(Schedulers.io())
            .subscribe(user -> {
                setUser(user);
                online.onNext(true);
                online.onComplete();
            },error -> {
                setUser(null);
                online.onNext(false);
                online.onComplete();
            });
    return online;
}

@Test
public void testGetStderrUpdatesEchoesUpdatesFromExecutorObservers() throws InterruptedException,ExecutionException,TimeoutException {
    final CancelablePromise<JobExecutionResult> executorPromise = new SimpleCancelablePromise<>();
    final Subject<byte[]> stderrSubject = PublishSubject.create();
    final JobExecutor executor =
            MockJobExecutor.thatUses(executorPromise,Observable.just(TestHelpers.generaterandomBytes()),stderrSubject);
    final JobManager jobManager = createManagerWith(executor);

    final Pair<JobId,CancelablePromise<FinalizedJob>> ret =
            jobManager.submit(STANDARD_VALID_REQUEST);

    final Observable<byte[]> stderrObservable =
            jobManager.stderrUpdates(ret.getLeft()).get();

    final atomicreference<byte[]> bytesFromObservable = new atomicreference<>();
    stderrObservable.subscribe(bytesFromObservable::set);

    final byte[] bytesExpected = TestHelpers.generaterandomBytes();
    stderrSubject.onNext(bytesExpected);

    executorPromise.complete(new JobExecutionResult(FINISHED));

    ret.getRight().get(DEFAULT_TIMEOUT,MILLISECONDS);

    assertthat(bytesFromObservable.get()).isEqualTo(bytesExpected);
}

private void initView() {
        subject = PublishSubject.create();
        subject.debounce(0,TimeUnit.MILLISECONDS)
//                .filter(new Predicate<Float>() {
//                    @Override
//                    public boolean test(Float contrast) throws Exception {
//                        return true;
//                    }
//                })
                .distinctUntilChanged()
                .switchMap(new Function<Float,ObservableSource<ColorMatrixColorFilter>>() {
                    @Override
                    public ObservableSource<ColorMatrixColorFilter> apply(Float value) throws Exception {
                        return postContrast(value);
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<ColorMatrixColorFilter>() {
                    @Override
                    public void accept(ColorMatrixColorFilter colorMatrixColorFilter) throws Exception {
                        setColorFilter(colorMatrixColorFilter);
                    }
                });
    }

public Observable<ScriptFile> download(String url,String path,MaterialDialog progressDialog) {
    PublishSubject<ScriptFile> subject = PublishSubject.create();
    DownloadManager.getInstance().download(url,path)
            .observeOn(AndroidSchedulers.mainThread())
            .doOnNext(progressDialog::setProgress)
            .subscribe(new SimpleObserver<Integer>() {
                @Override
                public void onComplete() {
                    progressDialog.dismiss();
                    subject.onNext(new ScriptFile(path));
                    subject.onComplete();
                }

                @Override
                public void onError(Throwable error) {
                    Log.e(LOG_TAG,"Download Failed",error);
                    progressDialog.dismiss();
                    showMessage(R.string.text_download_Failed);
                    subject.onError(error);
                }
            });
    return subject;
}

@Test
public void testPersistStderrReturnsAdisposableThatStopsFurtherReads() {
    final JobDAO dao = getInstance();
    final JobId jobId = dao.persist(STANDARD_VALID_REQUEST).getId();
    final Subject<byte[]> stderrSubject = PublishSubject.create();
    final AtomicBoolean stderrObsWasRead = new AtomicBoolean(false);
    final Observable<byte[]> stderrObs = stderrSubject.map(data -> {
        stderrObsWasRead.set(true);
        return data;
    });

    final disposable disposable = dao.appendStderr(jobId,stderrObs);
    disposable.dispose();
    stderrSubject.onNext(TestHelpers.generaterandomBytes());

    assertthat(stderrObsWasRead.get());
}

@Test
public void testExecuteStderrListenerIsCompletedOnceApplicationExecutionEnds() throws Throwable {
    final JobExecutor jobExecutor = getInstance();
    final AtomicBoolean completedCalled = new AtomicBoolean(false);
    final Subject<byte[]> stderrSubject = PublishSubject.create();
    stderrSubject.doOnComplete(() -> completedCalled.set(true)).subscribe();
    final JobEventListeners listeners = createStderrListener(stderrSubject);
    final CancelablePromise<JobExecutionResult> ret =
            jobExecutor.execute(STANDARD_REQUEST,listeners);

    promiseAssert(ret,result -> {
        try {
            // The stderr thread can race with the exit thread
            Thread.sleep(50);
            assertthat(completedCalled.get()).isTrue();
        } catch (InterruptedException ignored) {}
    });
}

@Test
public void testExecuteEvaluatesJobInputsAsExpected() throws InterruptedException {
    final JobExecutor jobExecutor = getInstance();
    final PersistedJob req =
            standardRequestWithCommand("echo","${inputs.foo}");
    final atomicreference<byte[]> bytesEchoedToStdout = new atomicreference<>(new byte[]{});
    final Subject<byte[]> stdoutSubject = PublishSubject.create();

    stdoutSubject.subscribe(bytes ->
            bytesEchoedToStdout.getAndUpdate(existingBytes ->
                    Bytes.concat(existingBytes,bytes)));

    final Semaphore s = new Semaphore(1);
    s.acquire();
    stdoutSubject.doOnComplete(s::release).subscribe();

    final JobEventListeners listeners =
            createStdoutListener(stdoutSubject);

    jobExecutor.execute(req,listeners);

    s.tryAcquire(TestConstants.DEFAULT_TIMEOUT,MILLISECONDS);

    final String stringFromStdout = new String(bytesEchoedToStdout.get()).trim();
    assertthat(stringFromStdout).isEqualTo("a"); // from spec
}

public static void main(String[] args) {
    Subject<String> subject = PublishSubject.create();
    subject.onNext("Alpha");
    subject.onNext("Beta");
    subject.onNext("Gamma");
    subject.onComplete();
    subject.map(String::length)
            .subscribe(System.out::println);
}

@TargetApi(Build.VERSION_CODES.M)
private Observable<Permission> requestImplementation(final String... permissions) {
    List<Observable<Permission>> list = new ArrayList<>(permissions.length);
    List<String> unrequestedPermissions = new ArrayList<>();

    // In case of multiple permissions,we create an Observable for each of them.
    // At the end,the observables are combined to have a unique response.
    for (String permission : permissions) {
        mRxPermissionsFragment.log("Requesting permission " + permission);
        if (isGranted(permission)) {
            // Already granted,or not Android M
            // Return a granted Permission object.
            list.add(Observable.just(new Permission(permission,true,false)));
            continue;
        }

        if (isRevoked(permission)) {
            // Revoked by a policy,return a denied Permission object.
            list.add(Observable.just(new Permission(permission,false)));
            continue;
        }

        PublishSubject<Permission> subject = mRxPermissionsFragment.getSubjectByPermission(permission);
        // Create a new subject if not exists
        if (subject == null) {
            unrequestedPermissions.add(permission);
            subject = PublishSubject.create();
            mRxPermissionsFragment.setSubjectForPermission(permission,subject);
        }

        list.add(subject);
    }

    if (!unrequestedPermissions.isEmpty()) {
        String[] unrequestedPermissionsArray = unrequestedPermissions.toArray(new String[unrequestedPermissions.size()]);
        requestPermissionsFromFragment(unrequestedPermissionsArray);
    }
    return Observable.concat(Observable.fromIterable(list));
}

public static void main(String[] args) {
    Subject<String> subject = PublishSubject.create();
    subject.map(String::length)
            .subscribe(System.out::println);
    subject.onNext("Alpha");
    subject.onNext("Beta");
    subject.onNext("Gamma");
    subject.onComplete();
}

public SelectedCountToolbarPresenter(Observable<Integer> selectedCountObservable,PublishSubject<Boolean> clearSelectionRelay,PublishSubject<Boolean> deleteSelectedItemsRelay) {
  this.selectedCountObservable = selectedCountObservable;
  this.clearSelectionRelay = clearSelectionRelay;
  this.deleteSelectedItemsRelay = deleteSelectedItemsRelay;
}

@Test
public void start_openDrawerIntent_shouldOpenTheDrawer() {
  // given
  TestObserver<SnackbarMessageEvent> snackbarMessage$ = new TestObserver<>();
  PublishSubject<Trigger> openDrawerIntent = PublishSubject.create();
  given(view.openDrawerIntent$()).willReturn(openDrawerIntent);
  given(view.readAboutIntent$()).willReturn(noTriggers());
  given(view.openProjectOnGitHubIntent$()).willReturn(noTriggers());
  given(view.selectLanguageIntent$()).willReturn(noTriggers());
  DrawerPresenter presenter = new DrawerPresenter(
      "http://foo.com",Sink.of(snackbarMessage$),urlOpener
  );
  presenter.start(view);

  // when
  fire(openDrawerIntent);

  // then
  verify(view).openDrawerIntent$();
  verify(view).readAboutIntent$();
  verify(view).openProjectOnGitHubIntent$();
  verify(view).selectLanguageIntent$();
  verify(view).openDrawer(true);
  then(view).shouldHaveNoMoreInteractions();
  then(urlOpener).shouldHaveZeroInteractions();
  snackbarMessage$.assertNovalues();
}

@Test
public void onEvent_presenterStopped_shouldIgnoreSubsequentEvent() {
  // given
  PublishSubject<String> subject = PublishSubject.create();
  TestPresenter presenter = new TestPresenter();
  presenter.on(subject).call(subscriber);
  presenter.stop();

  // when
  subject.onNext("foo");

  // then
  verifyZeroInteractions(subscriber);
}

@Override
public Searchviewmodel get(NavigationController navigationController,Observable<String> searchObservable,PublishSubject<SearchAdapter> searchAdapterSubject,SearchService searchService,SearchAdapterFactory searchAdapterFactory,Scheduler androidScheduler) {
    return new Searchviewmodel(navigationController,searchObservable,searchAdapterSubject,searchService,searchAdapterFactory,androidScheduler) {
        @Override
        public void search(String city) {
            latch.countDown();
        }
    };
}

@TargetApi(Build.VERSION_CODES.M)
private Observable<Permission> requestImplementation(Context context,final String... permissions) {
    List<Observable<Permission>> list = new ArrayList<>(permissions.length);
    List<String> unrequestedPermissions = new ArrayList<>();

    // In case of multiple permissions,the observables are combined to have a unique response.
    for (String permission : permissions) {
        XPermissionActivity.log("Requesting permission " + permission);
        if (isGranted(context,permission)) {
            // Already granted,false)));
            continue;
        }

        if (isRevoked(context,permission)) {
            // Revoked by a policy,false)));
            continue;
        }

        PublishSubject<Permission> subject = XPermissionActivity.getSubjectByPermission(permission);
        // Create a new subject if not exists
        if (subject == null) {
            unrequestedPermissions.add(permission);
            subject = PublishSubject.create();
            XPermissionActivity.setSubjectForPermission(permission,subject);
        }

        list.add(subject);
    }

    if (!unrequestedPermissions.isEmpty()) {
        String[] unrequestedPermissionsArray = unrequestedPermissions.toArray(new String[unrequestedPermissions.size()]);
        requestPermissionsFromActivity(context,unrequestedPermissionsArray);
    }
    return Observable.concat(Observable.fromIterable(list));
}

void onRequestPermissionsResult(String permissions[],size = permissions.length; i < size; i++) {
        log("onRequestPermissionsResult  " + permissions[i]);
        // Find the corresponding subject
        PublishSubject<Permission> subject = mSubjects.get(permissions[i]);
        if (subject == null) {
            // No subject found
            log("XPermission.onRequestPermissionsResult invoked but didn't find the corresponding permission request.");
            return;
        }
        mSubjects.remove(permissions[i]);
        boolean granted = grantResults[i] == PackageManager.PERMISSION_GRANTED;
        boolean showRequestPermissionRationale = shouldShowRequestPermissionRationale[i];
        log("granted: " + granted + "; showRequestPermissionRationale: " + showRequestPermissionRationale);
        if(Manifest.permission.WRITE_SETTINGS.equals(permissions[i]) || Manifest.permission.SYstem_ALERT_WINDOW.equals(permissions[i])){
            granted = PermissionsChecker.isPermissionGranted(this,permissions[i],false);
        }else{
            if(granted){
                if(PermissionsChecker.isPermissionGranted(this,true)){
                    granted = true;
                }else{
                    granted = false;
                    showRequestPermissionRationale = false;
                }
            }else if(showRequestPermissionRationale){
                if(PermissionsChecker.isPermissionGranted(this,false)){
                    granted = true;
                }else{
                    granted = false;
                }
            }
        }
        subject.onNext(new Permission(permissions[i],showRequestPermissionRationale));
        subject.onComplete();
    }
}

@Test
public void start_noEventPosted_shouldDonothingWithView() {
  // given
  PublishSubject<SnackbarMessageEvent> snackbarMessage$ = PublishSubject.create();
  SnackbarPresenter presenter = new SnackbarPresenter(snackbarMessage$);

  // when
  presenter.start(view);

  // then
  verifyZeroInteractions(view);
}

Store(@NonNull State initialState,@NonNull Reducer<State> reducer,@NonNull Effect<State>[] effects) {
    this.reducer = reducer;
    this.action$ = PublishSubject.create();
    this.state$ = BehaviorSubject.createDefault(initialState);
    this.result$ = Observable.fromArray(effects)
            .flatMap(transformer -> transformer.apply(action$,this::currentState));
}

/**
 * Subscribes the passed subject to the
 */
protected <T> void bindAction(Observable<T> from,PublishSubject<T> to) {
    composite.add(
            from
                    .doOnNext(t -> Timber.d("bind action called"))
                    .subscribe(to::onNext)
    );
}

@TargetApi(Build.VERSION_CODES.M)
private Observable<Permission> requestImplementation(final String... permissions) {
    List<Observable<Permission>> list = new ArrayList<>(permissions.length);
    List<String> unrequestedPermissions = new ArrayList<>();

    // In case of multiple permissions,the observables are combined to have a unique response.
    for (String permission : permissions) {
        Log.i("","Requesting permission " + permission);
        if (isGranted(permission)) {
            // Already granted,subject);
        }

        list.add(subject);
    }

    if (!unrequestedPermissions.isEmpty()) {
        String[] unrequestedPermissionsArray = unrequestedPermissions.toArray(new String[unrequestedPermissions.size()]);
        requestPermissionsFromFragment(unrequestedPermissionsArray);
    }
    return Observable.concat(Observable.fromIterable(list));
}

Dialogsstore(@NonNull AppStores base) {
    super(base);
    this.updatePublishSubject = PublishSubject.create();
    this.dialogsDeletingPublisher = PublishSubject.create();
    this.preferences = base.getSharedPreferences("dialogs_prefs",Context.MODE_PRIVATE);
    this.unreadDialogsCounter = PublishSubject.create();
}

@Test
public void testExecuteEvaluatesToJSONFunctionAsExpected() throws InterruptedException {
    final JobExecutor jobExecutor = getInstance();
    final PersistedJob req =
            standardRequestWithCommand("echo","${toJSON(inputs)}");
    final atomicreference<byte[]> bytesEchoedToStdout = new atomicreference<>(new byte[]{});
    final Subject<byte[]> stdoutSubject = PublishSubject.create();

    stdoutSubject.subscribe(bytes ->
            bytesEchoedToStdout.getAndUpdate(existingBytes ->
                    Bytes.concat(existingBytes,MILLISECONDS);

    final String stringFromStdout = new String(bytesEchoedToStdout.get()).trim();

    assertthat(stringFromStdout).isEqualTo(toJSON(STANDARD_VALID_REQUEST.getInputs()));
}

@Inject
public Messageviewmodel(MessageRepository messageRepository,@Named("vm") Compositedisposable compositedisposable) {
    Timber.d("Init Messageviewmodel");
    this.repository = messageRepository;
    this.compositedisposable = compositedisposable;
    this.messageIdentifier = new Identifier();
    this.subject = PublishSubject.create();
    this.messages = new mutablelivedata<>();
}

RealtimeMessagesProcessor() {
    this.app = Injection.provideApplicationContext();
    this.repositories = Injection.provideStores();
    this.networker = Injection.provideNetworkInterfaces();
    this.publishSubject = PublishSubject.create();
    this.queue = new LinkedList<>();
    this.notificationsInterceptors = new SparseArray<>(3);
    this.ownersInteractor = InteractorFactory.createOwnerInteractor();
    this.messagesInteractor = InteractorFactory.createMessagesInteractor();
}

/**
 * Download product from given category.
 *
 * @param request               which contains query category
 * @param productPublishSubject the subject which downloaded product should publish to
 */
public void downloadProducts(DownloadProductsRequest request,PublishSubject<Product> productPublishSubject) {

  QueryBuilder queryBuilder = QueryBuilders.termQuery("category",request.getCategory());
  SearchResponse scrollResponse =
      esClient
          .prepareSearch(INDEX)
          .setScroll(DEFAULT_SCROLL_TIME_VALUE)
          .setTypes(TYPE)
          .setQuery(queryBuilder)
          .setSize(SCROLL_SIZE)
          .get();
  do {
    scrollResponse.getHits().forEach(hit -> {
      try {
        Product.Builder builder = Product.newBuilder();
        jsonParser.merge(hit.sourceAsstring(),builder);
        productPublishSubject.onNext(builder.build());
      } catch (IOException ioe) {
        // Don't fail the whole stream
        log.error("Unable to read product record",ioe);
        productPublishSubject.onError(ioe);
        throw new IllegalStateException(ioe);
      }
    });
    // Fetch next batch of cite group records
    scrollResponse =
        esClient
            .prepareSearchScroll(scrollResponse.getScrollId())
            .setScroll(DEFAULT_SCROLL_TIME_VALUE)
            .execute()
            .actionGet();
  } while (scrollResponse.getHits().getHits().length != 0);

  productPublishSubject.onComplete();
}

@Test
public void calculateProductscore() throws Exception {
  PublishSubject<CalculateProductscoreResponse> publishSubject = PublishSubject.create();
  List<CalculateProductscoreResponse> responses = Lists.newArrayList();
  publishSubject
      .doOnNext(response -> responses.add(response))
      .subscribe();

  Product product = createProduct("category");
  productDao.calculateProductscore(product,publishSubject);

  assertthat(responses.size()).isEqualTo(1);
  publishSubject.onComplete();
}

/**
 * Subjects can be both observables and observers
 * Registration of the subscribers is important,a observer is only be notified if
 * a new events occurs
 * */
@Test
public void useSubject() {
    Subject<String> subject = PublishSubject.<String>create().toSerialized();
    subject.subscribe(e-> result+=e);
    subject.onNext("Hello");
    subject.onNext("Man");
    subject.onNext("Test");

    subject.subscribe(e-> doIt());
    subject.onNext("Test");
    assertthat(result).isEqualTo(_____);

}

/**
 * 接受消息
 *
 * @param tag      标志
 * @param callBack 回调
 */
public <T> disposableObserver register(@NonNull final Object tag,@NonNull final RxBusCallBack<T> callBack) {
    RxBusEvent rxBusEvent = rxBusEventArrayMap.get(tag);
    if (RxUtils.isEmpty(rxBusEvent)) {
        rxBusEvent = new RxBusEvent();
        rxBusEvent.subject = PublishSubject.create().toSerialized();
        rxBusEvent.disposable =
                rxBusEvent.subject
                        .ofType(callBack.busOfType())
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribeWith(new RxBusObserver<T>() {
                            @Override
                            public void onError(@io.reactivex.annotations.NonNull Throwable e) {
                                super.onError(e);
                                callBack.onBusError(e);
                            }

                            @Override
                            public void onNext(@io.reactivex.annotations.NonNull T t) {
                                super.onNext(t);
                                callBack.onBusNext(t);
                            }
                        });
        rxBusEventArrayMap.put(tag,rxBusEvent);
    }
    return rxBusEvent.disposable;
}

} *id;

...