package sia5;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.junit.Test;

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

public class FluxBufferingTests {

  @Test
  public void buffer() {
    Flux<String> fruitFlux = Flux.just(
        "jabłko", "pomarańcza", "banan", "kiwi", "truskawka");
    
    Flux<List<String>> bufferedFlux = fruitFlux.buffer(3);
    
    StepVerifier
        .create(bufferedFlux)
        .expectNext(Arrays.asList("jabłko", "pomarańcza", "banan"))
        .expectNext(Arrays.asList("kiwi", "truskawka"))
        .verifyComplete();
  }
  
  @Test
  public void bufferAndFlatMap() throws Exception {
    Flux.just(
        "jabłko", "pomarańcza", "banan", "kiwi", "truskawka")
        .buffer(3)
        .flatMap(x -> 
          Flux.fromIterable(x)
            .map(y -> y.toUpperCase())
            .subscribeOn(Schedulers.parallel())   
            .log()
        ).subscribe();
  }
  
  @Test
  public void collectList() {
    Flux<String> fruitFlux = Flux.just(
        "jabłko", "pomarańcza", "banan", "kiwi", "truskawka");
    
    Mono<List<String>> fruitListMono = fruitFlux.collectList();
    
    StepVerifier
        .create(fruitListMono)
        .expectNext(Arrays.asList(
            "jabłko", "pomarańcza", "banan", "kiwi", "truskawka"))
        .verifyComplete();
  }
  
  @Test
  public void collectMap() {
    Flux<String> animalFlux = Flux.just(
        "mrównik afrykański", "słoń", "koala", "orzeł", "kangur");
    
    Mono<Map<Character, String>> animalMapMono = 
        animalFlux.collectMap(a -> a.charAt(0));
    
    StepVerifier
        .create(animalMapMono)
        .expectNextMatches(map -> {
          return
              map.size() == 3 &&
              map.get('m').equals("mrównik afrykański") &&
              map.get('o').equals("orzeł") &&
              map.get('k').equals("kangur");
        })
        .verifyComplete();
  }
  
  @Test
  public void all() {
    Flux<String> animalFlux = Flux.just(
        "mrównik afrykański", "słoń", "koala", "orzeł", "kangur");
    
    Mono<Boolean> hasAMono = animalFlux.all(a -> a.contains("m"));
    StepVerifier.create(hasAMono)
      .expectNext(true)
      .verifyComplete();
    
    Mono<Boolean> hasKMono = animalFlux.all(a -> a.contains("o"));
    StepVerifier.create(hasKMono)
      .expectNext(false)
      .verifyComplete();
  }
  
  @Test
  public void any() {
    Flux<String> animalFlux = Flux.just(
        "mrównik afrykański", "słoń", "koala", "orzeł", "kangur");
    
    Mono<Boolean> hasAMono = animalFlux.any(a -> a.contains("m"));
    
    StepVerifier.create(hasAMono)
      .expectNext(true)
      .verifyComplete();
    
    Mono<Boolean> hasZMono = animalFlux.any(a -> a.contains("b"));
    StepVerifier.create(hasZMono)
      .expectNext(false)
      .verifyComplete();
  }
  
}
