Testing artefact for TDD with Rx Observables

TestRxStreamble.java

/*
 * Copyright (c) 2015.
 * David Navarro
 * me@davengeo.com
 */

public class TestRxStreamble implements Streamable<EventContainer> {

    private static Logger LOG = LoggerFactory.getLogger(TestRxStreamble.class);

    SettableFuture<EventContainer> future = SettableFuture.create();

    private Observable<EventContainer> stream = Observable.from(future, Schedulers.computation()).
            doOnEach(new Action1<Notification<? super EventContainer>>() {
                @Override
                public void call(Notification<? super EventContainer> notification) {
                    LOG.debug("value:{}", notification.getValue());
                    LOG.debug("next:{}, completed:{}, exception:{}",
                            notification.isOnNext(),
                            notification.isOnCompleted(),
                            notification.isOnError());

                }
            }).timeout(2, TimeUnit.SECONDS);

    @Override
    public Observable<EventContainer> stream() {
        return stream;
    }

    public void setEvent(EventContainer event) {
        future.set(event);
    }

    public void setError(Throwable error) {
        future.setException(error);
    }

    public boolean isDone() {
        return future.isDone();
    }

}