/*
 * Copyright © 2018-2019 Apple Inc. and the ServiceTalk project authors
 *
 * 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 io.servicetalk.transport.netty.internal;

import io.servicetalk.concurrent.api.Publisher;
import io.servicetalk.concurrent.api.TestPublisher;
import io.servicetalk.concurrent.api.TestPublisherSubscriber;
import io.servicetalk.transport.netty.internal.FlushStrategy.FlushSender;
import io.servicetalk.transport.netty.internal.NoopTransportObserver.NoopWriteObserver;

import io.netty.channel.EventLoop;
import org.junit.Test;

import static io.servicetalk.concurrent.api.SourceAdapters.toSource;
import static io.servicetalk.transport.netty.internal.Flush.composeFlushes;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.is;

public class FlushOutsideEventloopTest extends AbstractOutOfEventloopTest {

    private final TestPublisherSubscriber<Integer> subscriber = new TestPublisherSubscriber<>();
    private TestPublisher<Integer> src;
    private FlushSender flushSender;
    private MockFlushStrategy strategy;

    @Override
    public void setup0() {
        src = new TestPublisher<>();
        strategy = new MockFlushStrategy();
        Publisher<Integer> composedFlush = composeFlushes(channel, src, strategy, NoopWriteObserver.INSTANCE)
                .beforeOnNext(integer -> channel.write(integer));
        toSource(composedFlush).subscribe(subscriber);
        subscriber.request(Long.MAX_VALUE);
        flushSender = strategy.verifyApplied();
    }

    @Test
    public void testWriteAndFlushOutsideEventloop() throws Exception {
        EventLoop executor = getDifferentEventloopThanChannel();
        executor.submit(() -> src.onNext(1)).get();
        flushSender.flush();
        ensureEnqueuedTaskAreRun(executor);
        ensureEnqueuedTaskAreRun(channel.eventLoop());
        assertWritten(1);
    }

    @Test
    public void testWriteFromOutsideEventloopThenWriteAndFlushOnEventloop() throws Exception {
        EventLoop executor = getDifferentEventloopThanChannel();
        executor.submit(() -> src.onNext(1)).get();
        channel.eventLoop().submit(() -> {
            src.onNext(2);
            flushSender.flush();
        }).get();
        ensureEnqueuedTaskAreRun(executor);
        assertWritten(1, 2);
    }

    @Test
    public void testWriteInEventloopThenWriteAndFlushOutsideEventloop() throws Exception {
        EventLoop executor = getDifferentEventloopThanChannel();
        channel.eventLoop().submit(() -> src.onNext(1)).get();
        executor.submit(() -> {
            src.onNext(2);
            flushSender.flush();
        }).get();
        ensureEnqueuedTaskAreRun(executor);
        assertWritten(1, 2);
    }

    @Test
    public void testWriteFromEventloopAndFlushOutsideEventloop() throws Exception {
        EventLoop executor = getDifferentEventloopThanChannel();
        channel.eventLoop().submit(() -> src.onNext(1)).get();
        executor.submit(() -> flushSender.flush()).get();
        ensureEnqueuedTaskAreRun(executor);
        assertWritten(1);
    }

    private void assertWritten(Integer... values) throws InterruptedException {
        strategy.verifyWriteStarted();
        for (Integer v : values) {
            assertThat("Unexpected items written. Items pending flush: " + pendingFlush, written.take(), is(v));
        }
        strategy.verifyItemWritten(values.length);
        assertThat("Unexpected extra items written. Items pending flush: " + pendingFlush, written, is(empty()));
        src.onComplete();
        strategy.verifyWriteTerminated();
    }
}
