gistfile1.txt

import java.util.Arrays;
import java.util.List;

import junit.framework.TestCase;
import rx.Observable;
import rx.Observer;
import rx.subjects.PublishSubject;
import rx.util.functions.Func1;
import rx.util.functions.Func2;

public class RxToListTest extends TestCase {

	public void testToList() throws Exception {

		Dump observer = new Dump();
		PublishSubject<List<Integer>> subj = PublishSubject.create();

		subj.asObservable()
		// convert data from server to sequence
				.flatMap(new Func1<List<Integer>, Observable<List<String>>>() {

					@Override
					public Observable<List<String>> call(List<Integer> t1) {
						return Observable.from(t1).flatMap(new Func1<Integer, Observable<String>>() {

							@Override
							public Observable<String> call(Integer t1) {
								return Observable.combineLatest(Observable.from(t1), isInBookmarks(t1),
										new Func2<Integer, Boolean, String>() {

											@Override
											public String call(Integer t1, Boolean t2) {
												return t1 + " " + t2;
											}
										});

							}
						}).toList();
					}
				})
				// apply some transformation on every item (ie. which item is in bookmarks)

				// lets pass that to UI or
				.subscribe(observer);

		subj.onNext(Arrays.asList(1, 2));
		subj.onNext(Arrays.asList(3, 4));
		// subj.onCompleted();

		assertTrue(observer.nextCalled);
	}

	public Observable<Boolean> isInBookmarks(Integer item) {
		return Observable.from(item % 2 == 0);
	}

	class Dump implements Observer<Object> {

		boolean nextCalled;

		@Override
		public void onCompleted() {
			System.out.println("Obs " + " completed");
		}

		@Override
		public void onError(Throwable e) {
			System.out.println("Obs " + " error");
			e.printStackTrace();
		}

		@Override
		public void onNext(Object args) {
			nextCalled = true;
			System.out.println("Obs " + " next " + args.toString());
		}
	}
}