@Test
public void processingLongEffectsDoesNotBlockProcessingShorterEffects() {

  final List<String> effects = Arrays.asList("Hello", "Rx");
  PublishSubject<String> upstream = PublishSubject.create();
  Function<String, Integer> sleepyFunction =
      s -> {
        try {
          Thread.sleep(duration(s));
        } catch (InterruptedException ie) {
        }
        return s.length();
      };
  final List<Integer> results = new ArrayList<>();
  final Disposable disposable =
      upstream
          .compose(Transformers.fromFunction(sleepyFunction, Schedulers.io()))
          .subscribe(results::add);
  Observable.fromIterable(effects).subscribe(upstream);
  await().atMost(durationForEffects(effects)).until(() -> results.equals(expected(effects)));
  disposable.dispose();
}