Flux.fromIterable(getSomeLongList())
    .delayElements(Duration.ofMillis(100))
    .doOnNext(serviceA::someObserver)
    .map(d -> d * 2)
    .take(3)
    .onErrorResumeWith(errorHandler::fallback)
    .doAfterTerminate(serviceM::incrementTerminate)
    .subscribe(System.out::println);

Flux<String> empty = Flux.empty();
Flux<String> fromReadilyAvailableData = Flux.just("1", "2");
Flux<String> fromIterable = Flux.fromIterable(Arrays.asList("1", "2"));

Flux<string> fromError = Flux.error(new IllegalStateException("Hello"));

Flux<String> emits10IncreasingValuesAtRegularPace = Flux
  .interval(Duration.ofMillis(100))
  .take(10);

Mono.just(1)
    .map(integer -> "foo" + integer)
    .or(Mono.delay(Duration.ofMillis(100)))
    .subscribe(System.out::println);

StepVerifier.create(T<Publisher>).{expectations...}.verify();

StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofHours(3)))
            .expectSubscription()
            .expectNoEvent(Duration.ofHours(2))
            .thenAwait(Duration.ofHours(1))
            .expectNextCount(1)
            .expectComplete()
            .verify();

// TODO Create a StepVerifier
// that initially requests all values
// and expect 4 values to be received
StepVerifier requestAllExpectFour(Flux<User> flux) {
  return StepVerifier
    .create(flux)
    .expectNextCount(4)
    .expectComplete();
}

// TODO Create a StepVerifier
// that initially requests 1 value and expects User.SKYLER
// then requests another value and expects User.JESSE.
StepVerifier requestOneExpectSkylerThenRequestOneExpectJesse(Flux<User> flux) {
  return StepVerifier
    .create(flux)
    .expectNext(User.SKYLER)
    .expectNext(User.JESSE)
    .thenCancel(); // cancel하지 않으면 영원히 끝나지 않음.
}

// TODO Return a Flux
// with all users stored in the repository
// that prints automatically logs for all Reactive Streams signals
Flux<User> fluxWithLog() {
  return new ReactiveUserRepository()
    .findAll()
    .log();
}

// TODO Return a Flux with all users stored in the repository that prints "Starring:" on subscribe, "firstname lastname" for all values and "The end!" on complete
Flux<User> fluxWithDoOnPrintln() {
  return repository
    .findAll()
    .doOnSubscribe(s -> log.info("Starring:"))
    .doOnNext(u -> log.info(u.getFirstname() + " " + u.getLastname()))
    .doOnComplete(() -> log.info("The end!"));
}

/**
 * TODO Return a Mono<User> containing User.SAUL
 * when an error occurs in the input Mono,
 * else do not change the input Mono.
 */
Mono<User> betterCallSaulForBogusMono(Mono<User> mono) {
  return mono.onErrorReturn(User.SAUL);
}

/**
 * TODO Return a Flux<User> containing User.SAUL and User.JESSE
 * when an error occurs in the input Flux,
 * else do not change the input Flux.
 */
Flux<User> betterCallSaulAndJesseForBogusFlux(Flux<User> flux) {
  return .onErrorResume(e -> Flux.just(User.SAUL, User.JESSE));
}

/**
 * TODO Implement a method
 * that capitalizes each user of the incoming flux
 * using the #capitalizeUser method
 * and emits an error containing a GetOutOfHereException error
 */
Flux<User> capitalizeMany(Flux<User> flux) {
  return flux
    .map(u -> {
      try {
        return capitalizeUser(u);
      } catch (GetOutOfHereException e) {
        throw Exceptions.propaget(e);
      }
    });
}

// TODO Adapt Flux to RxJava Flowable
Flowable<User> fromFluxToFlowable(Flux<User> flux) {
    return Flowable.fromPublisher(flux);
}

// TODO Adapt RxJava Flowable to Flux
Flux<User> fromFlowableToFlux(Flowable<User> flowable) {
  return Flux.from(flowable);
}

// TODO Adapt Flux to RxJava Observable
Observable<User> fromFluxToObservable(Flux<User> flux) {
  return Observable.fromPublisher(flux);
}

// TODO Adapt RxJava Observable to Flux
Flux<User> fromObservableToFlux(Observable<User> observable) {
  return Flux.from(observable.toFlowable(BackpressureStrategy.BUFFER));
}

// Mono#doOnNext
public final Mono<T> doOnNext(Consumer<? super T> onNext) {
  Objects.requireNonNull(onNext, "onNext");
  return doOnSignal(this, (Consumer)null, onNext, (Consumer)null, (Runnable)null, (LongConsumer)null, (Runnable)null);
}

// MonoJust
MonoJust(T value) {
    this.value = Objects.requireNonNull(value, "value");
}

// TODO Create a Flux of user from Flux of username, firstname and lastname.
Flux<User> userFluxFromStringFlux(Flux<String> usernameFlux, Flux<String> firstnameFlux, Flux<String> lastnameFlux) {
  
  return Flux
      .zip(usernameFlux, firstnameFlux, lastnameFlux)
      .map(tuple3 -> new User(tuple3.getT1(), tuple3.getT2(), tuple3.getT3()));
}

// TODO Return the mono which returns its value faster
Mono<User> useFastestMono(Mono<User> mono1, Mono<User> mono2) {
  return Mono.first(mono1, mono2);
}

// TODO Return the flux which returns the first value faster
Flux<User> useFastestFlux(Flux<User> flux1, Flux<User> flux2) {
  return Flux.first(flux1, flux2);
}

// TODO Convert the input Flux<User> to a Mono<Void> that represents the complete signal of the flux
Mono<Void> fluxCompletion(Flux<User> flux) {
  return flux.ignoreElements().then();
}

// TODO Return a valid Mono of user for null input and non null input user (hint: Reactive Streams do not accept null values)
Mono<User> nullAwareUserToMono(User user) {
  return Mono.justOrEmpty(user);
}

// TODO Return the same mono passed as input parameter, expect that it will emit User.SKYLER when empty
Mono<User> emptyToSkyler(Mono<User> mono) {
  return mono.defaultIfEmpty(User.SKYLER);
}

@Test
public void fastestMono() {
  ReactiveRepository<User> repository = new ReactiveUserRepository(MARIE);
  ReactiveRepository<User> repositoryWithDelay = new ReactiveUserRepository(250, MIKE);
  Mono<User> mono = workshop.useFastestMono(repository.findFirst(), repositoryWithDelay.findFirst());
  StepVerifier.create(mono)
      .expectNext(MARIE)
      .verifyComplete();

  repository = new ReactiveUserRepository(250, MARIE);
  repositoryWithDelay = new ReactiveUserRepository(MIKE);
  mono = workshop.useFastestMono(repository.findFirst(), repositoryWithDelay.findFirst());
  StepVerifier.create(mono)
      .expectNext(MIKE)
      .verifyComplete();
}

// TODO Return the user contained in that Mono
User monoToValue(Mono<User> mono) {
  return mono.block();
}

// TODO Return the users contained in that Flux
Iterable<User> fluxToValues(Flux<User> flux) {
  return flux.toIterable();
}

// TODO Create a Flux for reading all users from the blocking repository deferred until the flux is subscribed, and run it with an elastic scheduler
Flux<User> blockingRepositoryToFlux(BlockingRepository<User> repository) {
  final Scheduler scheduler = Schedulers.elastic();

  return Flux
      .defer(() -> Flux.fromIterable(repository.findAll()))
      .subscribeOn(scheduler);
}

// TODO Insert users contained in the Flux parameter in the blocking repository using an elastic scheduler and return a Mono<Void> that signal the end of the operation
Mono<Void> fluxToBlockingRepository(Flux<User> flux, BlockingRepository<User> repository) {
  return flux
      .publishOn(Schedulers.elastic())
      .doOnNext(repository::save)
      .then();
}