OperationMaterialize { public static <T> OnSubscribeFunc<Notification<T>> materialize(final Observable<? extends T> sequence) { return new MaterializeObservable<T>(sequence); }  static OnSubscribeFunc<Notification<T>> materialize(final Observable<? extends T> sequence);  }
@Test public void testMaterialize1() { final TestAsyncErrorObservable o1 = new TestAsyncErrorObservable("one", "two", null, "three"); TestObserver Observer = new TestObserver(); Observable<Notification<String>> m = Observable.create(materialize(Observable.create(o1))); m.subscribe(Observer); try { o1.t.join(); } catch (InterruptedException e) { throw new RuntimeException(e); } assertFalse(Observer.onError); assertTrue(Observer.onCompleted); assertEquals(3, Observer.notifications.size()); assertEquals("one", Observer.notifications.get(0).getValue()); assertTrue(Observer.notifications.get(0).isOnNext()); assertEquals("two", Observer.notifications.get(1).getValue()); assertTrue(Observer.notifications.get(1).isOnNext()); assertEquals(NullPointerException.class, Observer.notifications.get(2).getThrowable().getClass()); assertTrue(Observer.notifications.get(2).isOnError()); }
@Test public void testMaterialize2() { final TestAsyncErrorObservable o1 = new TestAsyncErrorObservable("one", "two", "three"); TestObserver Observer = new TestObserver(); Observable<Notification<String>> m = Observable.create(materialize(Observable.create(o1))); m.subscribe(Observer); try { o1.t.join(); } catch (InterruptedException e) { throw new RuntimeException(e); } assertFalse(Observer.onError); assertTrue(Observer.onCompleted); assertEquals(4, Observer.notifications.size()); assertEquals("one", Observer.notifications.get(0).getValue()); assertTrue(Observer.notifications.get(0).isOnNext()); assertEquals("two", Observer.notifications.get(1).getValue()); assertTrue(Observer.notifications.get(1).isOnNext()); assertEquals("three", Observer.notifications.get(2).getValue()); assertTrue(Observer.notifications.get(2).isOnNext()); assertTrue(Observer.notifications.get(3).isOnCompleted()); }
@Test public void testMultipleSubscribes() throws InterruptedException, ExecutionException { final TestAsyncErrorObservable o = new TestAsyncErrorObservable("one", "two", null, "three"); Observable<Notification<String>> m = Observable.create(materialize(Observable.create(o))); assertEquals(3, m.toList().toBlockingObservable().toFuture().get().size()); assertEquals(3, m.toList().toBlockingObservable().toFuture().get().size()); }