/**
 * Copyright (c) 2016 European Organisation for Nuclear Research (CERN), All Rights Reserved.
 */

package cern.streaming.pool.core.service.streamfactory;

import static cern.streaming.pool.core.service.util.ReactiveStreams.fromRx;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.assertj.core.api.Assertions.assertThat;

import org.junit.Before;
import org.junit.Test;

import cern.streaming.pool.core.service.StreamId;
import cern.streaming.pool.core.service.streamfactory.CombineWithLatestStreamFactory;
import cern.streaming.pool.core.service.streamid.CombineWithLatestStreamId;
import cern.streaming.pool.core.service.util.ReactiveStreams;
import cern.streaming.pool.core.testing.subscriber.BlockingTestSubscriber;
import rx.Observable;

/**
 * Unit tests for {@link CombineWithLatestStreamFactory}
 * 
 * @author acalia
 */
public class CombineWithLatestStreamIdStreamFactoryTest {

    private CombineWithLatestStreamFactory factory;
    private BlockingTestSubscriber<Long> subscriber;

    @Before
    public void setUp() {
        factory = new CombineWithLatestStreamFactory();
        subscriber = BlockingTestSubscriber.ofName("Subscriber");
    }

    /* @formatter:off
     * Trigger +--------------T--------------T--------------T--------------T--------------T-------------------->
     * Data    +---------0---------1---------2---------3---------4---------5---------6---------7---------8----->
     *
     * Result  +--------------0--------------2--------------3--------------5--------------6-------------------->
     * @formatter:on
     */
    @Test
    public void test1() {
        Observable<Long> trigger = Observable.interval(1500, MILLISECONDS).take(5);
        Observable<Long> data = Observable.interval(1000, MILLISECONDS);

        subscribeAndWait(data, trigger);

        assertThat(subscriber.getValues()).containsExactly(0L, 2L, 3L, 5L, 6L);
    }

    /* @formatter:off
     * Trigger +----T------------------------T--------------T--------------T-------------------T--------------->
     * Data    +---------0---------1---------2---------3---------4---------5---------6---------7---------8----->
     *
     * Result  +-----------------------------2--------------3--------------5-------------------7--------------->
     * @formatter:on
     */
    @Test
    public void test2() {
        Observable<Long> trigger = Observable.merge(delayed(500), delayed(3000), delayed(4500), delayed(6500),
                delayed(8000));
        Observable<Long> data = Observable.interval(1000, MILLISECONDS);

        subscribeAndWait(data, trigger);

        assertThat(subscriber.getValues()).containsExactly(2L, 3L, 5L, 7L);
    }

    /* @formatter:off
     * Trigger +----T-------->
     * Data    +---------0--->
     *
     * Result  +------------->
     * @formatter:on
     */
    @Test
    public void test3() {
        Observable<Long> trigger = delayed(500);
        Observable<Long> data = delayed(1000);

        subscribeAndWait(data, trigger);

        assertThat(subscriber.getValues()).isEmpty();
    }

    /* @formatter:off
     * Trigger +----T------------------------------------------------------------------------------------------>
     * Data    +---------0---------1---------2---------3---------4---------5---------6---------7---------8----->
     *
     * Result  +----------------------------------------------------------------------------------------------->
     * @formatter:on
     */
    @Test
    public void test4() {
        Observable<Long> trigger = delayed(500);
        Observable<Long> data = Observable.interval(1000, MILLISECONDS);
        subscribeAndWait(data, trigger);
        assertThat(subscriber.getValues()).isEmpty();
    }

    private void subscribeAndWait(Observable<Long> data, Observable<Long> trigger) {
        CombineWithLatestStreamId streamId = CombineWithLatestStreamId.of(fromRx(data), fromRx(trigger));
        ReactiveStreams.publisherFrom(factory.create(streamId, null)).subscribe(subscriber);
        subscriber.await();
    }

    private Observable<Long> delayed(int millis) {
        return Observable.just(-1L).delay(millis, MILLISECONDS);
    }

}

@@ -1240,32 +1240,37 @@
 ------T---------
+T----
 -----T----------
@@ -1260,37 +1260,32 @@
 ------T---------
------
 T--------------T
@@ -1275,33 +1275,32 @@
 -T--------------
-T
 --------------T-
@@ -1289,34 +1289,32 @@
 ----------------
-T-
 ----------------
@@ -1418,35 +1418,32 @@
 ---7---------8--
----
 %3E%0A     *%0A     * 
@@ -1471,36 +1471,35 @@
 --0---------
--
+1
 ----
-2
 -----
--
+2
 --------3---
@@ -1501,33 +1501,32 @@
 -3--------------
-5
 --------------6-
@@ -1515,33 +1515,32 @@
 ----------------
-6
 ----------------
@@ -1666,16 +1666,41 @@
 terval(1
+000, MILLISECONDS).delay(
 500, MIL
@@ -1715,17 +1715,17 @@
 S).take(
-5
+4
 );%0A     
@@ -1901,23 +1901,19 @@
 (0L,
+ 1L,
  2L, 3L
-, 5L, 6L
 );%0A 
@@ -1977,32 +1977,37 @@
 ----------------
+-----
 T--------------T
@@ -1992,38 +1992,38 @@
 ------T---------
+T
 -----
-T
 --------------T-
@@ -2043,37 +2043,32 @@
 -----T----------
------
 %3E%0A     * Data   
@@ -2214,30 +2214,30 @@
 ------------
-2
 -----
+2
 ---------3--
@@ -2240,32 +2240,37 @@
 -3--------------
+-----
 5---------------
@@ -2276,37 +2276,32 @@
 -----7----------
------
 %3E%0A     * @format
@@ -2426,17 +2426,17 @@
 elayed(3
-0
+5
 00), del
@@ -2487,17 +2487,17 @@
 elayed(8
-0
+5
 00));%0A  

fix in test