/*
 * Copyright 2016-present David Karnok
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package hu.akarnokd.rxjava3.operators;

import static org.junit.Assert.assertFalse;

import java.io.IOException;

import org.junit.Test;

import hu.akarnokd.rxjava3.test.*;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.functions.BiConsumer;
import io.reactivex.rxjava3.internal.functions.Functions;
import io.reactivex.rxjava3.operators.QueueFuseable;
import io.reactivex.rxjava3.processors.*;

public class FlowableMapFilterConditionalTest {

    @Test
    public void map() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult(2, 4, 6, 8, 10);
    }

    @Test
    public void take() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .take(3)
        .test()
        .assertResult(2, 4, 6);
    }

    @Test
    public void filter() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                if (t % 2 == 0) {
                    e.doNext(t * 2);
                }
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult(4, 8);
    }

    @Test
    public void mapAndComplete() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
                e.doComplete();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult(2);
    }

    @Test
    public void mapTwice() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IllegalStateException.class, 2);
    }

    @Test
    public void mapHidden() {
        Flowable.range(1, 5).hide()
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult(2, 4, 6, 8, 10);
    }

    @Test
    public void filterHidden() {
        Flowable.range(1, 5).hide()
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                if (t % 2 == 0) {
                    e.doNext(t * 2);
                }
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult(4, 8);
    }

    @Test
    public void consumerThrows() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                throw new IOException();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IOException.class);
    }

    @Test
    public void consumerSignalsError() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doError(new IOException());
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IOException.class);
    }

    @Test
    public void consumerSignalsErrorCancel() {
        BehaviorProcessor<Integer> pp = BehaviorProcessor.createDefault(1);

        pp
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doError(new IOException());
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IOException.class);

        assertFalse(pp.hasSubscribers());
    }

    @Test
    public void consumerThrowsCancel() {
        BehaviorProcessor<Integer> pp = BehaviorProcessor.createDefault(1);

        pp
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                throw new IOException();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IOException.class);

        assertFalse(pp.hasSubscribers());
    }

    @Test
    public void consumerCompletes() {
        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doComplete();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult();
    }

    @Test
    public void consumerCompleteCancel() {
        BehaviorProcessor<Integer> pp = BehaviorProcessor.createDefault(1);

        pp
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doComplete();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertResult();

        assertFalse(pp.hasSubscribers());
    }

    @Test
    public void mapFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.SYNC)
        .assertResult(2, 4, 6, 8, 10);
    }

    @Test
    public void mapAsyncFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        UnicastProcessor<Integer> up = UnicastProcessor.create();
        TestHelper.emit(up, 1, 2, 3, 4, 5);

        up
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doNext(t * 2);
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.ASYNC)
        .assertResult(2, 4, 6, 8, 10);
    }

    @Test
    public void filterFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                if (t % 2 == 0) {
                    e.doNext(t * 2);
                }
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.SYNC)
        .assertResult(4, 8);
    }

    @Test
    public void consumerThrowsFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                throw new IOException();
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.SYNC)
        .assertFailure(IOException.class);
    }

    @Test
    public void consumerSignalsErrorFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doError(new IOException());
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.SYNC)
        .assertFailure(IOException.class);
    }

    @Test
    public void consumerCompleteFused() {
        TestSubscriberEx<Integer> ts = TestHelper.fusedSubscriber(QueueFuseable.ANY);

        Flowable.range(1, 5)
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doComplete();
            }
        }))
        .filter(Functions.alwaysTrue())
        .subscribe(ts);

        ts.assertFusionMode(QueueFuseable.SYNC)
        .assertResult();
    }

    @Test
    public void error() {
        Flowable.<Integer>error(new IOException())
        .compose(FlowableTransformers.mapFilter(new BiConsumer<Integer, BasicEmitter<Integer>>() {
            @Override
            public void accept(Integer t, BasicEmitter<Integer> e) throws Exception {
                e.doComplete();
            }
        }))
        .filter(Functions.alwaysTrue())
        .test()
        .assertFailure(IOException.class);
    }
}
