public static class Cylon {}

public static class Jail {

Object cylon;

Jail(Object cylon) {

this.cylon = cylon;

}

}

public static class CylonDetectorObservable<T> {

protected OnSubscribe<T> onSubscribe;

public static <T> CylonDetectorObservable<T> create(OnSubscribe<T> onSubscribe) {

return new CylonDetectorObservable<T>(onSubscribe);

}

protected CylonDetectorObservable(OnSubscribe<T> onSubscribe) {

this.onSubscribe = onSubscribe;

}

public void subscribe(Subscriber<T> subscriber) {

onSubscribe.call(subscriber);

}

public <R> CylonDetectorObservable<R> lift(Operator<? extends R, ? super T> operator) {

return x(new RobotConversionFunc<T, R>(operator));

}

public <R, O> O x(Func1<OnSubscribe<T>, O> operator) {

return operator.call(onSubscribe);

}

public <R> CylonDetectorObservable<? extends R> compose(Func1<CylonDetectorObservable<? super T>, CylonDetectorObservable<? extends R>> transformer) {

return transformer.call(this);

}

public final CylonDetectorObservable<T> beep(Func1<? super T, Boolean> predicate) {

return lift(new OperatorFilter<T>(predicate));

}

public final <R> CylonDetectorObservable<R> boop(Func1<? super T, ? extends R> func) {

return lift(new OperatorMap<T, R>(func));

}

public CylonDetectorObservable<String> DESTROY() {

return boop(new Func1<T, String>() {

@Override

public String call(T t) {

Object cylon = ((Jail) t).cylon;

throwOutTheAirlock(cylon);

if (t instanceof Jail) {

String name = cylon.toString();

return "Cylon '" + name + "' has been destroyed";

}

else {

return "Cylon 'anonymous' has been destroyed";

}

}});

}

private static void throwOutTheAirlock(Object cylon) {

// ...

}

}

public static class RobotConversionFunc<T, R> implements Func1<OnSubscribe<T>, CylonDetectorObservable<R>> {

private Operator<? extends R, ? super T> operator;

public RobotConversionFunc(Operator<? extends R, ? super T> operator) {

this.operator = operator;

}

@Override

public CylonDetectorObservable<R> call(final OnSubscribe<T> onSubscribe) {

return CylonDetectorObservable.create(new OnSubscribe<R>() {

@Override

public void call(Subscriber<? super R> o) {

try {

Subscriber<? super T> st = operator.call(o);

try {

st.onStart();

onSubscribe.call(st);

} catch (OnErrorNotImplementedException e) {

throw e;

} catch (Throwable e) {

st.onError(e);

}

} catch (OnErrorNotImplementedException e) {

throw e;

} catch (Throwable e) {

o.onError(e);

}

}});

}

}

public static class ConvertToCylonDetector<T> implements Func1<OnSubscribe<T>, CylonDetectorObservable<T>> {

@Override

public CylonDetectorObservable<T> call(final OnSubscribe<T> onSubscribe) {

return CylonDetectorObservable.create(onSubscribe);

}

}

public static class ConvertToObservable<T> implements Func1<OnSubscribe<T>, Observable<T>> {

@Override

public Observable<T> call(final OnSubscribe<T> onSubscribe) {

return Observable.create(onSubscribe);

}

}

@Test

public void testConversionBetweenObservableClasses() {

final TestSubscriber<String> subscriber = new TestSubscriber<String>(new Subscriber<String>(){

@Override

public void onCompleted() {

System.out.println("Complete");

}

@Override

public void onError(Throwable e) {

System.out.println("error: " + e.getMessage());

e.printStackTrace();

}

@Override

public void onNext(String t) {

System.out.println(t);

}});

List<Object> crewOfBattlestarGalactica = Arrays.asList(new Object[] {"William Adama", "Laura Roslin", "Lee Adama", new Cylon()});

Observable.from(crewOfBattlestarGalactica)

.extend(new ConvertToCylonDetector<Object>())

.beep(new Func1<Object, Boolean>(){

@Override

public Boolean call(Object t) {

return t instanceof Cylon;

}})

.boop(new Func1<Object, Object>() {

@Override

public Jail call(Object cylon) {

return new Jail(cylon);

}})

.DESTROY()

.x(new ConvertToObservable<String>())

.reduce("Cylon Detector finished. Report:\n", new Func2<String, String, String>() {

@Override

public String call(String a, String n) {

return a + n + "\n";

}})

.subscribe(subscriber);

subscriber.assertNoErrors();

subscriber.assertCompleted();

}

@Test

public void testConvertToConcurrentQueue() {

final AtomicReference<Throwable> thrown = new AtomicReference<Throwable>(null);

final AtomicBoolean isFinished = new AtomicBoolean(false);

ConcurrentLinkedQueue<? extends Integer> queue = Observable.range(0,5)

.flatMap(new Func1<Integer, Observable<Integer>>(){

@Override

public Observable<Integer> call(final Integer i) {

return Observable.range(0, 5)

.observeOn(Schedulers.io())

.map(new Func1<Integer, Integer>(){

@Override

public Integer call(Integer k) {

try {

Thread.sleep(System.currentTimeMillis() % 100);

} catch (InterruptedException e) {

e.printStackTrace();

}

return i + k;

}});

}})

.extend(new Func1<OnSubscribe<Integer>, ConcurrentLinkedQueue<Integer>>() {

@Override

public ConcurrentLinkedQueue<Integer> call(OnSubscribe<Integer> onSubscribe) {

final ConcurrentLinkedQueue<Integer> q = new ConcurrentLinkedQueue<Integer>();

onSubscribe.call(new Subscriber<Integer>(){

@Override

public void onCompleted() {

isFinished.set(true);

}

@Override

public void onError(Throwable e) {

thrown.set(e);

}

@Override

public void onNext(Integer t) {

q.add(t);

}});

return q;

}});

int x = 0;

while(!isFinished.get()) {

Integer i = queue.poll();

if (i != null) {

x++;

System.out.println(x + " item: " + i);

}

}

assertEquals(null, thrown.get());

}