/*
 * Copyright © 2014-2017 EntIT Software LLC, a Micro Focus company (L.P.)
 *
 * 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.cloudslang.worker.management.services;

import io.cloudslang.orchestrator.entities.Message;
import io.cloudslang.orchestrator.services.OrchestratorDispatcherService;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatcher;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

import static org.junit.Assert.assertEquals;
import static org.mockito.Matchers.anyList;
import static org.mockito.Matchers.anyString;
import static org.mockito.Matchers.argThat;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.verify;


@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class OutboundBufferTest {

    private final Logger logger = LogManager.getLogger(getClass());

    private static final int MAX_BUFFER_WEIGHT = 10;
    private static final int MAX_BULK_WEIGHT = 3;

    @Autowired
    private WorkerRecoveryManager recoveryManager;

    @Autowired
    private OutboundBuffer buffer;

    @Autowired
    private OrchestratorDispatcherService dispatcherService;

    @Before
    public void setUp() {
        ((WorkerRecoveryListener) buffer).doRecovery();
        reset(recoveryManager, dispatcherService);
    }

    /**
     * Makes sure the buffer aggregates messages and dispatches them in bulk
     */
    @Test
    public void testAggregation() throws InterruptedException {
        List<DummyMsg1> messages = Arrays.asList(new DummyMsg1(), new DummyMsg1());

        for (DummyMsg1 message : messages) {
            buffer.put(message);
        }

        buffer.drain();
        verify(dispatcherService)
                .dispatch((List<? extends Serializable>) argThat(new MessagesSizeMatcher(messages)), anyString(),
                        anyString(), anyString());
    }

    /**
     * checks that when inserting messages to a full buffer, the inserting thread will block until the buffer is
     * emptied
     */
    @Test
    public void testProducerBlocking() throws InterruptedException {
        // buffer capacity is 10, put messages in it until it is full
        while (buffer.getWeight() < MAX_BUFFER_WEIGHT) {
            buffer.put(new DummyMsg1());
        }

        // the next insert to buffer will block because it's full, do it on a different thread
        Thread thread = new Thread(() -> {
			try {
				buffer.put(new DummyMsg1());
			} catch (InterruptedException e) {
				//ignore
			}
		});
        thread.start();

        // wait for that thread to block
        waitForThreadStateToBe(thread, Thread.State.WAITING);
        assertEquals("inserting thread should be in a waiting state when inserting to full buffer",
                Thread.State.WAITING, thread.getState());

        // drain the buffer -> will send the first 10 messages and release the blocking thread
        buffer.drain();
        waitForThreadStateToBe(thread, Thread.State.TERMINATED);
        assertEquals("inserting thread should be in a terminated state after inserting to buffer",
                Thread.State.TERMINATED, thread.getState());
        thread.join();
    }

    /**
     * Checks that the buffer will block when having no messages and continues when the first message arrives
     */
    @Test
    public void testConsumerBlocking() throws InterruptedException {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                buffer.drain();
            }
        });
        thread.start();

        // draining the buffer should block since it is empty
        waitForThreadStateToBe(thread, Thread.State.WAITING);
        assertEquals("reading thread should be in a waiting state when inserting to full buffer", Thread.State.WAITING,
                thread.getState());

        // insert 2 new messages
        Message[] messages = new Message[]{new DummyMsg1(), new DummyMsg2()};
        buffer.put(messages);

        // thread should be released now
        waitForThreadStateToBe(thread, Thread.State.TERMINATED);
        assertEquals("reading thread should be in a terminated after a message was inserted to the buffer",
                Thread.State.TERMINATED, thread.getState());

        thread.join();
        verify(dispatcherService)
                .dispatch((List<? extends Serializable>) argThat(new MessagesSizeMatcher(Arrays.asList(messages))),
                        anyString(), anyString(), anyString());
    }


    private void waitForThreadStateToBe(Thread thread, Thread.State state) throws InterruptedException {
        int waitCount = 0;
        while (!thread.getState().equals(state) && waitCount <= 20) {
            Thread.sleep(50);
            waitCount++;
        }
    }

    @Test
    public void longevityTest() throws InterruptedException {
        int THREADS_NUM = 5;
        long CHECK_DURATION = 5 * 1000L;
        long INFO_FREQUENCY = 2 * 1000L;

        final AtomicBoolean run = new AtomicBoolean(true);
        final CountDownLatch latch = new CountDownLatch(THREADS_NUM + 1);

        for (int i = 1; i <= THREADS_NUM; i++) {
            final int index = i;
            new Thread(new Runnable() {
                private final Class<? extends Message> messageClass =
                        (index % 2) != 0 ? DummyMsg1.class : DummyMsg2.class;

                @Override
                public void run() {
                    int counter = 0;
                    try {
                        logger.debug("started, will generate messages of " + messageClass.getSimpleName());

                        while (run.get()) {
                            buffer.put(messageClass.newInstance());
                            counter++;
                            Thread.sleep(5L);
                        }
                        logger.debug("thread finished. processed " + counter + " messages");
                    } catch (Exception ex) {
                        logger.error("thread finished", ex);
                    } finally {
                        latch.countDown();
                    }
                }
            }, "T-" + i).start();
        }

        final DrainStatistics statistics = new DrainStatistics();
        //noinspection unchecked
        doAnswer((Answer<Object>) invocation -> {
			@SuppressWarnings("unchecked") List<Message> messages = (List<Message>) invocation.getArguments()[0];
			int weight = 0;
			for (Message message : messages) {
				weight += message.getWeight();
			}
			statistics.add(messages.size(), weight);
			return null;
		}).when(dispatcherService).dispatch(anyList(), anyString(), anyString(), anyString());

        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    logger.debug("started");

                    while (run.get()) {
                        buffer.drain();
                        Thread.sleep(30L);
                    }

					while (buffer.getSize() > 0) {
						buffer.drain();
					}
                } catch (Exception ex) {
                    logger.error("thread finished", ex);
                } finally {
                    latch.countDown();
                }
            }
        }, "T-D").start();

        long t = System.currentTimeMillis();
        while (System.currentTimeMillis() - t < CHECK_DURATION) {
            Thread.sleep(INFO_FREQUENCY);
            logger.debug(buffer.getStatus());
        }
        run.set(false);
        latch.await();

        System.out.println("Drain statistics: " + statistics.report());
    }


    /**
     * Makes sure the recovery clears worker state
     */
    @Test
    public void testRecovery() throws InterruptedException {
        List<DummyMsg1> messages = Arrays.asList(new DummyMsg1(), new DummyMsg1());

        for (DummyMsg1 message : messages) {
            buffer.put(message);
        }
        assertEquals(1, buffer.getSize());
        assertEquals(2, buffer.getWeight());
        ((WorkerRecoveryListener) buffer).doRecovery();
        assertEquals(0, buffer.getSize());
        assertEquals(0, buffer.getWeight());
    }

    private class MessagesSizeMatcher extends ArgumentMatcher {

        List messages;

        public MessagesSizeMatcher(List val) {
            messages = val;
        }

        @Override
        public boolean matches(Object argument) {
            if (argument instanceof List) {
                List listConverted = (List) argument;
                if (listConverted.size() == messages.size()) {
                    return true;
                }
            }
            return false;
        }
    }

    static class DrainStatistics {

        private int counter;
        private int size;
        private int weight;

        public void add(int size, int weight) {
            counter++;
            this.size += size;
            this.weight += weight;
        }

        public String report() {
            return "Buffer has sent " + counter + " bulks, avg(size): " + size / counter + ", avg(weight): "
                    + weight / counter + ", total messages: " + size;
        }
    }


    static class DummyMsg1 implements Message {

        public int getWeight() {
            return 1;
        }

        public String getId() {
            return "";
        }

        public List<Message> shrink(List<Message> messages) {
            return messages;
        }
    }

    static class DummyMsg2 implements Message {

        public int getWeight() {
            return 2;
        }

        public String getId() {
            return "";
        }

        public List<Message> shrink(List<Message> messages) {
            return messages;
        }
    }

    @Configuration
    static class config {

        static {
            System.setProperty("out.buffer.max.buffer.weight", String.valueOf(MAX_BUFFER_WEIGHT));
            System.setProperty("out.buffer.max.bulk.weight", String.valueOf(MAX_BULK_WEIGHT));
        }

        @Bean
        public WorkerRecoveryManager workerRecoveryManager() {
            return mock(WorkerRecoveryManager.class);
        }

        @Bean
        OrchestratorDispatcherService orchestratorDispatcherService() {
            return mock(OrchestratorDispatcherService.class);
        }

        @Bean
        SynchronizationManager synchronizationManager() {
            return new SynchronizationManagerImpl();
        }

        @Bean
        public RetryTemplate retryTemplate() {
            return new RetryTemplate();
        }

        @Bean
        public OutboundBuffer outboundBuffer() {
            return new OutboundBufferImpl();
        }

        @Bean
        public String workerUuid() {
            return "1234";
        }

        @Bean
        public Integer numberOfExecutionThreads() {
            return 10;
        }

    }

}
