package sia5;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

import org.junit.Test;

import lombok.Data;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

public class FluxTransformingTests {

  @Test
  public void skipAFew() {
    Flux<String> countFlux = Flux.just(
        "jeden", "dwa", "pomiń kilka", "dziewięćdziesiąt dziewięć", "sto")
        .skip(3);
   
    StepVerifier.create(countFlux)
        .expectNext("dziewięćdziesiąt dziewięć", "sto")
        .verifyComplete();
  }
  
  @Test
  public void skipAFewSeconds() {
    Flux<String> countFlux = Flux.just(
        "jeden", "dwa", "pomiń kilka", "dziewięćdziesiąt dziewięć", "sto")
        .delayElements(Duration.ofSeconds(1))
        .skip(Duration.ofSeconds(4));
   
    StepVerifier.create(countFlux)
        .expectNext("dziewięćdziesiąt dziewięć", "sto")
        .verifyComplete();
  }
  
  @Test
  public void take() {
    Flux<String> nationalParkFlux = Flux.just(
        "Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Wielki Kanion Kolorado", "Park Narodowy Zion", "Park Narodowy Acadia")
        .take(3);
   
    StepVerifier.create(nationalParkFlux)
        .expectNext("Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Wielki Kanion Kolorado")
        .verifyComplete();
  }
  
  @Test
  public void takeForAwhile() {
    Flux<String> nationalParkFlux = Flux.just(
        "Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Wielki Kanion Kolorado", "Park Narodowy Zion", "Park Narodowy Grand Teton")
        .delayElements(Duration.ofSeconds(1))
        .take(Duration.ofMillis(3500));
   
    StepVerifier.create(nationalParkFlux)
        .expectNext("Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Wielki Kanion Kolorado")
        .verifyComplete();
  }
  
  @Test
  public void filter() {
    Flux<String> nationalParkFlux = Flux.just(
        "Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Wielki Kanion Kolorado", "Park Narodowy Zion", "Park Narodowy Grand Teton")
        .filter(np -> !np.contains(" "));
   
    StepVerifier.create(nationalParkFlux)
        .expectNext("Park Narodowy Yellowstone", "Park Narodowy Yosemite", "Park Narodowy Zion")
        .verifyComplete();
  }
  
  @Test
  public void distinct() {
    Flux<String> animalFlux = Flux.just(
        "pies", "kot", "ptak", "pies", "ptak", "mrówkojad")
        .distinct();
   
    StepVerifier.create(animalFlux)
        .expectNext("pies", "kot", "ptak", "mrówkojad")
        .verifyComplete();
  }
  
  @Test
  public void map() {
    Flux<Player> playerFlux = Flux
      .just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
      .map(n -> {
        String[] split = n.split("\\s");
        return new Player(split[0], split[1]);
      });
    
    StepVerifier.create(playerFlux)
        .expectNext(new Player("Michael", "Jordan"))
        .expectNext(new Player("Scottie", "Pippen"))
        .expectNext(new Player("Steve", "Kerr"))
        .verifyComplete();
  }
  
  @Test
  public void flatMap() {
    Flux<Player> playerFlux = Flux
      .just("Michael Jordan", "Scottie Pippen", "Steve Kerr")
      .flatMap(n -> Mono.just(n)
          .map(p -> {
              String[] split = p.split("\\s");
              return new Player(split[0], split[1]);
            })
          .subscribeOn(Schedulers.parallel())
        );
    
    List<Player> playerList = Arrays.asList(
        new Player("Michael", "Jordan"), 
        new Player("Scottie", "Pippen"), 
        new Player("Steve", "Kerr"));

    StepVerifier.create(playerFlux)
        .expectNextMatches(p -> playerList.contains(p))
        .expectNextMatches(p -> playerList.contains(p))
        .expectNextMatches(p -> playerList.contains(p))
        .verifyComplete();
  }
  
  @Data
  private static class Player {
    private final String firstName;
    private final String lastName;
  }
  
}
