是否存在只传播错误而不终止自身的可观察性？

Question

我在负责同步的类中使用PublishSubject。当同步完成时，将通知所有订阅服务器。发生错误时也会发生同样的情况。我注意到，下次我在发生错误后订阅时，它会立即返回给订阅者。

因此类可能如下所示：

public class Synchronizer {
private final PublishSubject<Result> mSyncHeadObservable = PublishSubject.create();
    private final ThreadPoolExecutor mExecutor = new ThreadPoolExecutor(1, 1,
            10, TimeUnit.SECONDS,
            new SynchronousQueue<Runnable>(true),
            new ThreadPoolExecutor.DiscardPolicy());


    public Observable<Result> syncHead(final int chunkSize) {
              mExecutor.execute(new Runnable() {
      @Override
      public void run() {
          try {
              //Do some work which either returns a result or throws an error
              //...

              mSyncHeadObservable.onNext(Notification.createOnNext(/*some result*/));
          } catch (Throwable error) {
              mSyncHeadObservable.onError(Notification.<Result>createOnError(error));
          }
      }
  });

是否有一个可观察的东西可以作为代理？可能是其他Rx方法？

更新：我遵循了@Akarnokd方法，发出包装到RxJava通知中的事件。然后通过flatmap()展开它们。因此synchronizer类的客户端不需要执行此操作。

    //...
private PublishSubject<Notification<Result>> mSyncHeadObservable = PublishSubject.create();

      public Observable<Result> syncHead(final int chunkSize) {

          return mSyncHeadObservable.flatMap(new Func1<Notification<Result>, Observable<Result>>() {
            @Override
            public Observable<Result> call(Notification<Result> result) {
                if (result.isOnError()) {
                    return Observable.error(result.getThrowable());
                }

                return Observable.just(result.getValue());
            }
        }).doOnSubscribe(
            new Action0() {
                @Override
                public void call() {
                    startHeadSync(chunkSize);
                }
            });
      }

      private void startHeadSync(final int chunkSize) {
          mExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    //Do some work which either returns a result or throws an error
                    //...

                    mSyncHeadObservable.onNext(Notification.createOnNext(/*some result*/));
                } catch (Throwable error) {
                    mSyncHeadObservable.onError(Notification.<Result>createOnError(error));
                }
            }
          });
       }
  //...

Answer 1

我不确定您希望通过这种设置实现什么，但通常，为了避免PublishSubject出现终端条件，您应该将值和错误封装到一个公共结构中，并始终发出这些值和错误，永远不要发出任何onerror和onCompleted。一种选择是使用RXJava自己的事件包装器通知，您的订阅服务器应该解压缩该值。