PublishSubject extends Subject<T> { public void post(T value) { notifyObservers(value); }  void post(T value);  }
@Test public void TestValue() { PublishSubject<String> observable = new PublishSubject<>(); observable.post("1"); Disposable subscriptionA = observable.subscribe(new Observer<String>() { @Override public void onChanged(String value) { addResult("A" + value); } }); AssertResults(); observable.post("2"); AssertResults("A2"); Disposable subscriptionB = observable.subscribe(new Observer<String>() { @Override public void onChanged(String value) { addResult("B" + value); } }); AssertResults(); observable.post("3"); AssertResults("A3", "B3"); subscriptionA.dispose(); observable.post("4"); AssertResults("B4"); subscriptionB.dispose(); observable.post("5"); AssertResults(); observable.subscribe(new Observer<String>() { @Override public void onChanged(String value) { addResult("C" + value); } }); AssertResults(); observable.post("6"); AssertResults("C6"); }
@Test public void TestRemoveObserver() { Observer<String> observer = new Observer<String>() { @Override public void onChanged(String value) { addResult(value); } }; PublishSubject<String> observable = new PublishSubject<>(); observable.subscribe(observer); observable.post("1"); AssertResults("1"); observable.removeObserver(observer); observable.post("2"); AssertResults(); }
@Test public void TestRemoveObserverWhileNotifying() { PublishSubject<String> observable = new PublishSubject<>(); final Disposable subscriptionA = observable.subscribe(new Observer<String>() { @Override public void onChanged(String value) { addResult("A" + value); } }); observable.subscribe(new Observer<String>() { @Override public void onChanged(String value) { addResult("B" + value); subscriptionA.dispose(); } }); observable.post("1"); AssertResults("A1", "B1"); observable.post("2"); AssertResults("B2"); }