在传递覆盖onSubscribe的订阅服务器时调用dispose（）

Question

@NonNull Observable<Long> src = Observable.interval(1, TimeUnit.SECONDS);
src.subscribe(new Observer<Long>() {
      private Disposable d;

      @Override
      public void onSubscribe(@NonNull Disposable d) {
           this.d = d;
      }

      @Override
      public void onNext(@NonNull Long aLong) {
           if(!d.isDisposed()) {
              System.out.println("Number onNext = " + aLong);
           }
      }

       @Override
       public void onError(@NonNull Throwable e) {

       }

       @Override
       public void onComplete() {
           System.out.println("completed");
       }
 });

但我不知道如何为该订阅调用dispose()。subscribe通过observer作为参数返回void并且subscribeWith在没有编译错误的情况下不接受我的observer。

这怎么行？我误会了什么？

Answer 1

@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
    ...
}

@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {
    ...
}

// Create a reference of disposable object
Disposable d = Observable.interval(1, TimeUnit.SECONDS)
            .subscribe(longTime -> {
                System.out.println("Number onNext = " + aLong);
            }, throwable -> {
                
            })

// dispose it
d.dispose()