Subject extends Observable<T> implements Observer<T> { protected Subject() { } protected  Subject(); @NonNull static Subject<T> create(); @Override synchronized void onNext(@NonNull T value); @Override synchronized void onCompleted(); @Override synchronized void onError(@NonNull Exception e); @NonNull @Override synchronized Subscription subscribe(@NonNull final Observer<T> observer);  }
@Test public void testSubject() throws Exception { Subject<Integer> subject = Subject.create(); subject.subscribe(new Subscriber<Integer>() { @Override public void onNext(@NonNull Integer value) { Assert.assertEquals(value.intValue(), 3); resultMap.put("next", true); } @Override public void onCompleted() { resultMap.put("complete", true); } @Override public void onError(@NonNull Exception e) { resultMap.put("error", true); } }); subject.onNext(3); subject.onCompleted(); Assert.assertTrue(resultMap.get("next")); Assert.assertTrue(resultMap.get("complete")); Assert.assertFalse(resultMap.get("error")); }