code: @Override @CallerSideMethod
      public void subscribe(Subscriber<? super OUT> subscriber) {
          if ( isRemote() ) {
              synchronized (this) { // subscribe/unsubscribe won't be contended
                  if ( _callerSideSubscribers == null ) {
                      _callerSideSubscribers = new ArrayList();
                  }
                  _callerSideSubscribers.add(subscriber);
              }
          }
          if ( subscriber == null )
          {
              subscriber.onError( new IllegalArgumentException("cannot subscibe null") );
              return;
          }
          _subscribe((res, err) -> {
              if (isError(err)) {
                  subscriber.onError((Throwable) err);
              } else if (isComplete(err)) {
                  subscriber.onComplete();
              } else {
                  subscriber.onNext((OUT) res);
              }
          }).then(subs -> {
              Log.Info(this, "stream subscribe acknowledged");
              subscriber.onSubscribe(subs);
          });
      }
docs: advanced chemistry: held+used in the remote proxy to clean up pipeline on disconnect