/*
 * Copyright 2012-2018 Chronicle Map Contributors
 *
 * 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 net.openhft.chronicle.map;

import net.openhft.chronicle.bytes.Bytes;
import net.openhft.chronicle.hash.replication.ReplicableEntry;
import net.openhft.chronicle.threads.NamedThreadFactory;
import org.hamcrest.CoreMatchers;
import org.junit.Test;

import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;

public class BasicReplicationTest {

    private static byte asByte(final int i) {
        return (byte) i;
    }

    @Test
    public void shouldReplicate() {
        final ChronicleMapBuilder<String, String> builder = ChronicleMap.of(String.class, String.class)
                .entries(1000).averageKeySize(7).averageValueSize(7);
        try (
                ReplicatedChronicleMap<String, String, Object> mapOne = createReplicatedMap(builder, asByte(1));
                ReplicatedChronicleMap<String, String, Object> mapTwo = createReplicatedMap(builder, asByte(2));
                ReplicatedChronicleMap<String, String, Object> mapThree = createReplicatedMap(builder, asByte(3))
        ) {

            final ReplicationEventProcessor<String, String> processorOne =
                    new ReplicationEventProcessor<>();
            processorOne.addDestinationMap(mapOne.acquireModificationIterator(asByte(2)), mapOne, mapTwo);
            processorOne.addDestinationMap(mapOne.acquireModificationIterator(asByte(3)), mapOne, mapThree);

            final ReplicationEventProcessor<String, String> processorTwo =
                    new ReplicationEventProcessor<>();
            processorTwo.addDestinationMap(mapTwo.acquireModificationIterator(asByte(1)), mapTwo, mapOne);
            processorTwo.addDestinationMap(mapTwo.acquireModificationIterator(asByte(3)), mapTwo, mapThree);

            final ReplicationEventProcessor<String, String> processorThree =
                    new ReplicationEventProcessor<>();
            processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(1)), mapThree, mapOne);
            processorThree.addDestinationMap(mapThree.acquireModificationIterator(asByte(2)), mapThree, mapTwo);

            final ExecutorService executorService = Executors.newFixedThreadPool(3,
                    new NamedThreadFactory("test"));
            executorService.submit(processorOne::processPendingChangesLoop);
            executorService.submit(processorTwo::processPendingChangesLoop);
            executorService.submit(processorThree::processPendingChangesLoop);

            final List<Map<String, String>> maps = Arrays.asList(mapOne, mapTwo, mapThree);
            final Random random = new Random(0xBAD5EED);
            for (int i = 0; i < 5000; i++) {
                final int mapIndex = random.nextInt(maps.size());
                final Map<String, String> map = maps.get(mapIndex);
                final String key = "key" + random.nextInt(100);
                final String value = "val" + random.nextInt(500);
                map.put(key, value);
            }

            waitForBacklog(processorOne);
            waitForBacklog(processorTwo);
            waitForBacklog(processorThree);

            executorService.shutdownNow();

            waitForFinish(processorOne);
            waitForFinish(processorTwo);
            waitForFinish(processorThree);

            assertThat(mapOne.size(), is(equalTo(mapTwo.size())));
            assertThat(mapOne.size(), is(equalTo(mapThree.size())));

            for (String key : mapOne.keySet()) {
                final String mapOneValue = mapOne.get(key);
                final String mapTwoValue = mapTwo.get(key);
                final String mapThreeValue = mapThree.get(key);

                assertThat(mapOneValue, CoreMatchers.equalTo(mapTwoValue));
                assertThat(mapOneValue, CoreMatchers.equalTo(mapThreeValue));
            }
        }
    }

    private void waitForBacklog(final ReplicationEventProcessor<String, String> p) {
        while (!p.queueEmpty.get()) {
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
        }
    }

    private void waitForFinish(final ReplicationEventProcessor<String, String> processor) {
        for (IteratorAndDestinationMap<String, String> destinationMap : processor.destinationMaps) {
            while (destinationMap.messagesInflight.get() != 0 || !processor.stopped.get()) {
                LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(1L));
            }
        }
    }

    private ReplicatedChronicleMap<String, String, Object>
    createReplicatedMap(final ChronicleMapBuilder<String, String> builder, final byte replicaId) {
        final ChronicleMap<String, String> map = builder.replication(replicaId).create();
        return (ReplicatedChronicleMap<String, String, Object>) map;
    }

    private static final class ReplicationEventProcessor<K, V> {
        private final List<IteratorAndDestinationMap<K, V>> destinationMaps = new ArrayList<>();
        private final AtomicBoolean queueEmpty = new AtomicBoolean(true);
        private final AtomicBoolean stopped = new AtomicBoolean(false);
        private ReplicatedChronicleMap<K, V, ?> sourceMap;

        void addDestinationMap(final ReplicatedChronicleMap<K, V, ?>.ModificationIterator modificationIterator,
                               final ReplicatedChronicleMap<K, V, ?> sourceMap,
                               final ReplicatedChronicleMap<K, V, ?> destinationMap) {

            destinationMaps.add(new IteratorAndDestinationMap<>(modificationIterator, sourceMap, destinationMap));
            if (this.sourceMap != null && this.sourceMap != sourceMap) {
                throw new IllegalArgumentException("All iterators must belong to the same source map");
            }
            this.sourceMap = sourceMap;
            modificationIterator.setModificationNotifier(this::wakeup);
        }

        void wakeup() {
            queueEmpty.set(false);
        }

        void processPendingChangesLoop() {

            try {

                while (!Thread.currentThread().isInterrupted()) {
                    while (queueEmpty.get() && !Thread.currentThread().isInterrupted()) {
                        LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(7));
                    }

                    Collections.shuffle(destinationMaps);

                    for (IteratorAndDestinationMap<K, V> iteratorAndDestinationMap : destinationMaps) {
                        while (iteratorAndDestinationMap.modificationIterator.nextEntry(
                                iteratorAndDestinationMap, sourceMap.identifier())) {
                        }
                    }

                    queueEmpty.compareAndSet(false, true);
                }
            } finally {
                stopped.set(true);
            }
        }
    }

    private static final class IteratorAndDestinationMap<K, V> implements Replica.ModificationIterator.Callback {
        private final ReplicatedChronicleMap<K, V, ?>.ModificationIterator modificationIterator;
        private final ReplicatedChronicleMap<K, V, ?> sourceMap;
        private final ReplicatedChronicleMap<K, V, ?> destinationMap;
        private final Bytes<ByteBuffer> buffer = Bytes.elasticByteBuffer(4096);
        private final ExecutorService delayedExecutor = Executors.newSingleThreadExecutor(
                new NamedThreadFactory("delayed"));
        private final AtomicInteger messagesInflight = new AtomicInteger(0);
        private final ArrayList<String> keys = new ArrayList<>();

        IteratorAndDestinationMap(final ReplicatedChronicleMap<K, V, ?>.ModificationIterator modificationIterator,
                                  final ReplicatedChronicleMap<K, V, ?> sourceMap,
                                  final ReplicatedChronicleMap<K, V, ?> destinationMap) {
            this.modificationIterator = modificationIterator;
            this.sourceMap = sourceMap;
            this.destinationMap = destinationMap;
        }

        @Override
        public void onEntry(final ReplicableEntry entry, final int chronicleId) {
            try {
                buffer.clear();
                keys.clear();
                sourceMap.writeExternalEntry(entry, null, buffer, chronicleId, keys);

                buffer.readPosition(0);
                buffer.readLimit(buffer.writePosition());
                final ByteBuffer message = ByteBuffer.allocate((int) buffer.writePosition());
                buffer.read(message);
                message.position(0);
                messagesInflight.incrementAndGet();
                delayedExecutor.submit(() -> {
                    try {
                        final Bytes<ByteBuffer> tmp = Bytes.elasticByteBuffer(128);
                        while (message.remaining() != 0) {
                            tmp.writeByte(message.get());
                        }

                        destinationMap.readExternalEntry(tmp, sourceMap.identifier());
                        messagesInflight.decrementAndGet();
                    } catch (Throwable e) {
                        e.printStackTrace();
                    }
                });
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }

        @Override
        public void onBootstrapTime(final long bootstrapTime, final int chronicleId) {
            destinationMap.setRemoteNodeCouldBootstrapFrom((byte) chronicleId, bootstrapTime);
        }
    }
}
