2.x: sync Javadoc of Observable (copy&adjust Flowable's) (#4346)

Author: akarnokd

src/main/java/io/reactivex/Completable.java

@@ -837,7 +837,7 @@ public final Completable endWith(CompletableSource other) {
      */
     @SchedulerSupport(SchedulerSupport.CUSTOM)
     public final <T> Observable<T> endWith(ObservableSource<T> next) {
-        return this.<T>toObservable().endWith(next);
+        return this.<T>toObservable().concatWith(next);
     }
 
     /**
@@ -1102,7 +1102,7 @@ public final Completable startWith(CompletableSource other) {
     @SchedulerSupport(SchedulerSupport.NONE)
     public final <T> Observable<T> startWith(Observable<T> other) {
         Objects.requireNonNull(other, "other is null");
-        return other.endWith(this.<T>toObservable());
+        return other.concatWith(this.<T>toObservable());
     }
     /**
      * Returns an Observable which first delivers the events

src/main/java/io/reactivex/Observable.java

@@ -46,7 +46,7 @@ public ObservableAutoConnect(ConnectableObservable<? extends T> source,
 
     @Override
     public void subscribeActual(Observer<? super T> child) {
-        source.unsafeSubscribe(child);
+        source.subscribe(child);
         if (clients.incrementAndGet() == numberOfSubscribers) {
             source.connect(connection);
         }

src/main/java/io/reactivex/internal/operators/observable/ObservableAutoConnect.java

@@ -192,7 +192,7 @@ public void onSubscribe(Disposable s) {
          * Make sure this is called only once.
          */
         public void connect() {
-            source.unsafeSubscribe(this);
+            source.subscribe(this);
             isConnected = true;
         }
         @Override

src/main/java/io/reactivex/internal/operators/observable/ObservableCache.java

@@ -111,7 +111,7 @@ void doSubscribe(final Observer<? super T> subscriber, final CompositeDisposable
 
         ConnectionSubscriber s = new ConnectionSubscriber(subscriber, currentBase, d);
 
-        source.unsafeSubscribe(s);
+        source.subscribe(s);
     }
 
     private Disposable disconnect(final CompositeDisposable current) {

src/main/java/io/reactivex/internal/operators/observable/ObservableRefCount.java

@@ -780,7 +780,7 @@ public void onComplete() {
             }
         };
 
-        Observable.range(1, 1).delay(1, TimeUnit.SECONDS).buffer(2, TimeUnit.SECONDS).unsafeSubscribe(s);
+        Observable.range(1, 1).delay(1, TimeUnit.SECONDS).buffer(2, TimeUnit.SECONDS).subscribe(s);
 
         cdl.await();
 

src/test/java/io/reactivex/internal/operators/observable/ObservableBufferTest.java

@@ -255,7 +255,7 @@ public void onNext(Integer t) {
             }
         };
 
-        source.unsafeSubscribe(ts);
+        source.subscribe(ts);
 
         Assert.assertEquals(100, count.get());
 

src/test/java/io/reactivex/internal/operators/observable/ObservableCacheTest.java

@@ -384,7 +384,7 @@ public Observable<Integer> apply(Integer t1) {
         Observer<Object> o = TestHelper.mockObserver();
         InOrder inOrder = inOrder(o);
 
-        source.delay(subFunc, delayFunc).subscribe(o);
+        source.delay(Observable.defer(subFunc), delayFunc).subscribe(o);
 
         source.onNext(1);
         delay.onNext(1);
@@ -419,7 +419,7 @@ public Observable<Integer> apply(Integer t1) {
         Observer<Object> o = TestHelper.mockObserver();
         InOrder inOrder = inOrder(o);
 
-        source.delay(subFunc, delayFunc).subscribe(o);
+        source.delay(Observable.defer(subFunc), delayFunc).subscribe(o);
 
         source.onNext(1);
         delay.onNext(1);
@@ -453,7 +453,7 @@ public Observable<Integer> apply(Integer t1) {
         Observer<Object> o = TestHelper.mockObserver();
         InOrder inOrder = inOrder(o);
 
-        source.delay(subFunc, delayFunc).subscribe(o);
+        source.delay(Observable.defer(subFunc), delayFunc).subscribe(o);
 
         source.onNext(1);
         delay.onError(new TestException());
@@ -513,7 +513,7 @@ public Observable<Integer> apply(Integer t1) {
         Observer<Object> o = TestHelper.mockObserver();
         InOrder inOrder = inOrder(o);
 
-        source.delay(subFunc, delayFunc).subscribe(o);
+        source.delay(Observable.defer(subFunc), delayFunc).subscribe(o);
 
         source.onNext(1);
         sdelay.onComplete();
@@ -733,13 +733,8 @@ public Integer apply(Integer t) {
     public void testBackpressureWithSelectorDelayAndSubscriptionDelay() {
         TestObserver<Integer> ts = new TestObserver<Integer>();
         Observable.range(1, Flowable.bufferSize() * 2)
-                .delay(new Callable<Observable<Long>>() {
-
-                    @Override
-                    public Observable<Long> call() {
-                        return Observable.timer(500, TimeUnit.MILLISECONDS);
-                    }
-                }, new Function<Integer, Observable<Long>>() {
+                .delay(Observable.timer(500, TimeUnit.MILLISECONDS)
+                , new Function<Integer, Observable<Long>>() {
 
                     @Override
                     public Observable<Long> apply(Integer i) {
@@ -800,12 +795,7 @@ public void testDelaySupplierSimple() {
 
         TestObserver<Integer> ts = new TestObserver<Integer>();
 
-        source.delaySubscription(new Callable<Observable<Integer>>() {
-            @Override
-            public Observable<Integer> call() {
-                return ps;
-            }
-        }).subscribe(ts);
+        source.delaySubscription(ps).subscribe(ts);
 
         ts.assertNoValues();
         ts.assertNoErrors();
@@ -826,12 +816,7 @@ public void testDelaySupplierCompletes() {
 
         TestObserver<Integer> ts = new TestObserver<Integer>();
 
-        source.delaySubscription(new Callable<Observable<Integer>>() {
-            @Override
-            public Observable<Integer> call() {
-                return ps;
-            }
-        }).subscribe(ts);
+        source.delaySubscription(ps).subscribe(ts);
 
         ts.assertNoValues();
         ts.assertNoErrors();
@@ -853,12 +838,7 @@ public void testDelaySupplierErrors() {
 
         TestObserver<Integer> ts = new TestObserver<Integer>();
 
-        source.delaySubscription(new Callable<Observable<Integer>>() {
-            @Override
-            public Observable<Integer> call() {
-                return ps;
-            }
-        }).subscribe(ts);
+        source.delaySubscription(ps).subscribe(ts);
 
         ts.assertNoValues();
         ts.assertNoErrors();

src/test/java/io/reactivex/internal/operators/observable/ObservableDelayTest.java

@@ -109,7 +109,7 @@ public void testHonorsContractWhenCompleted() {
 
         Observer<Integer> o = TestHelper.mockObserver();
 
-        result.unsafeSubscribe(o);
+        result.subscribe(o);
 
         verify(o).onNext(1);
         verify(o).onComplete();
@@ -124,7 +124,7 @@ public void testHonorsContractWhenThrows() {
 
         Observer<Integer> o = TestHelper.mockObserver();
 
-        result.unsafeSubscribe(o);
+        result.subscribe(o);
 
         verify(o, never()).onNext(any(Integer.class));
         verify(o, never()).onComplete();

src/test/java/io/reactivex/internal/operators/observable/ObservableDematerializeTest.java

@@ -32,7 +32,7 @@ public void before() {
     }
 
     private void checkActionCalled(Observable<String> input) {
-        input.finallyDo(aAction0).subscribe(NbpObserver);
+        input.doAfterTerminate(aAction0).subscribe(NbpObserver);
         verify(aAction0, times(1)).run();
     }