package com.balamaci.rx;

import com.balamaci.rx.util.Pair;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.flowables.GroupedFlowable;
import org.junit.Test;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author sbalamaci
 */
public class Part05AdvancedOperators implements BaseTestObservables {

    @Test
    public void buffer() {
        Flowable<Long> numbers = Flowable.interval(1, TimeUnit.SECONDS);

        Flowable<List<Long>> delayedNumbersWindow = numbers
                .buffer(5);

        subscribeWithLog(delayedNumbersWindow);
    }

    @Test
    public void simpleWindow() {
        Flowable<Long> numbers = Flowable.interval(1, TimeUnit.SECONDS);

        Flowable<Long> delayedNumbersWindow = numbers
                .window(5)
                .flatMap(window -> window.doOnComplete(() -> log.info("Window completed")));

        subscribeWithLog(delayedNumbersWindow);
    }


    @Test
    public void window() {
        Flowable<Long> numbers = Flowable.interval(1, TimeUnit.SECONDS);

        Flowable<Long> delayedNumbersWindow = numbers
                .window(10, 5, TimeUnit.SECONDS)
                .flatMap(window -> window.doOnComplete(() -> log.info("Window completed")));

        subscribeWithLog(delayedNumbersWindow);
    }

    /**
     * groupBy splits the stream into multiple streams with the key generated by the function passed as
     * parameter to groupBy
     */
    @Test
    public void groupBy() {
        Flowable<String> colors = Flowable.fromArray("red", "green", "blue",
                "red", "yellow", "green", "green");

        Flowable<GroupedFlowable<String, String>> groupedColorsStream = colors
                .groupBy(val -> val); //identity function
//                .groupBy(val -> "length" + val.length());

        Flowable<Pair<String, Long>> colorCountStream = groupedColorsStream
                .flatMap(groupedColor -> groupedColor
                                            .count()
                                            .map(count -> new Pair<>(groupedColor.getKey(), count))
                                            .toFlowable()
                );

        subscribeWithLog(colorCountStream);
    }

    @Test
    public void bufferWithLimitTriggeredByObservable() {
        Observable<String> colors = Observable.fromArray("red", "green", "blue",
                "red", "yellow", "#", "green", "green");


        colors.publish(p -> p.filter(val -> ! val.equals("#"))
                             .buffer(() -> p.filter(val -> val.equals("#")))
                )
                .subscribe(list -> {
            String listCommaSeparated = String.join(",", list);

            log.info("List {}", listCommaSeparated);
        });
    }
}
