/*
 * Copyright 2019, TeamDev. All rights reserved.
 *
 * Redistribution and use in source and/or binary forms, with or without
 * modification, must retain the above copyright notice and the following
 * disclaimer.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package io.spine.server.delivery.given;

import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
import com.google.protobuf.Any;
import io.spine.protobuf.AnyPacker;
import io.spine.server.NodeId;
import io.spine.server.aggregate.AggregateRepository;
import io.spine.server.delivery.InboxMessage;
import io.spine.server.delivery.ShardIndex;
import io.spine.server.delivery.ShardObserver;
import io.spine.server.route.EventRoute;
import io.spine.server.route.EventRouting;

import java.security.SecureRandom;
import java.util.Map;

import static io.spine.base.Identifier.newUuid;

/**
 * Test environment for {@link Delivery} tests.
 */
public class DeliveryTestEnv {

    /** Prevents instantiation of this utility class. */
    private DeliveryTestEnv() {
    }

    public static ImmutableSet<String> manyTargets(int size) {
        return new SecureRandom().ints(size)
                                 .mapToObj((i) -> "calc-" + i)
                                 .collect(ImmutableSet.toImmutableSet());
    }

    public static ImmutableSet<String> singleTarget() {
        return ImmutableSet.of("the-calculator");
    }

    public static NodeId generateNodeId() {
        return NodeId
                .newBuilder()
                .setValue(newUuid())
                .vBuild();
    }

    public static ShardIndex newShardIndex(int index, int ofTotal) {
        return ShardIndex
                .newBuilder()
                .setIndex(index)
                .setOfTotal(ofTotal)
                .vBuild();
    }

    public static class CalculatorRepository extends AggregateRepository<String, CalcAggregate> {

        /** How many calls there were to {@link #doStore(CalcAggregate)} method, grouped by ID. */
        private static final Map<String, Integer> storeCalls = Maps.newConcurrentMap();

        /** How many calls there were to {@link #doLoadOrCreate(String)} method, grouped by ID. */
        private static final Map<String, Integer> loadOrCreateCalls = Maps.newConcurrentMap();

        public CalculatorRepository() {
            storeCalls.clear();
            loadOrCreateCalls.clear();
        }

        @Override
        protected void setupImportRouting(EventRouting<String> routing) {
            routeByFirstField(routing);
        }

        @Override
        protected void setupEventRouting(EventRouting<String> routing) {
            routeByFirstField(routing);
        }

        private static void routeByFirstField(EventRouting<String> routing) {
            routing.replaceDefault(EventRoute.byFirstMessageField(String.class));
        }

        @Override
        protected void doStore(CalcAggregate aggregate) {
            String id = aggregate.id();
            incrementByKey(id, storeCalls);
            super.doStore(aggregate);
        }

        @Override
        protected CalcAggregate doLoadOrCreate(String id) {
            incrementByKey(id, loadOrCreateCalls);
            return super.doLoadOrCreate(id);
        }

        private static void incrementByKey(String id, Map<String, Integer> counters) {
            if (!counters.containsKey(id)) {
                counters.put(id, 0);
            }
            counters.put(id, counters.get(id) + 1);
        }

        public int storeCallsCount(String id) {
            return storeCalls.getOrDefault(id, 0);
        }

        public int loadOrCreateCallsCount(String id) {
            return loadOrCreateCalls.getOrDefault(id, 0);
        }
    }

    /**
     * An observer of the messages delivered to shards, which remembers all the delivered messages
     * per {@code CalculatorAggregate} identifier.
     *
     * <p>Unpacks {@linkplain InboxMessage#getPayloadCase() the payload} of each
     * {@code InboxMessage} observed.
     */
    public static class SignalMemoizer implements ShardObserver {

        private final Multimap<String, CalculatorSignal> signals = ArrayListMultimap.create();

        @Override
        public synchronized void onMessage(InboxMessage update) {
            Any packed;
            if (update.hasCommand()) {
                packed = update.getCommand()
                               .getMessage();
            } else {
                packed = update.getEvent()
                               .getMessage();

            }
            CalculatorSignal msg =
                    (CalculatorSignal) AnyPacker.unpack(packed);
            signals.put(msg.getCalculatorId(), msg);

        }

        public ImmutableSet<CalculatorSignal> messagesBy(String id) {
            return ImmutableSet.copyOf(signals.get(id));
        }
    }

    /**
     * An observer of the messages delivered to shards, which remembers all the delivered messages.
     *
     * <p>Memoizes the observed {@code InboxMessage}s as-is.
     */
    public static class RawMessageMemoizer implements ShardObserver {

        private final ImmutableList.Builder<InboxMessage> rawMessages = ImmutableList.builder();

        @Override
        public void onMessage(InboxMessage update) {
            rawMessages.add(update);
        }

        public ImmutableList<InboxMessage> messages() {
            return rawMessages.build();
        }
    }

    /**
     * An observer of the messages delivered to shards, which remembers all the shards involved.
     */
    public static class ShardIndexMemoizer implements ShardObserver {

        private final ImmutableSet.Builder<ShardIndex> observedShards = ImmutableSet.builder();

        @Override
        public void onMessage(InboxMessage update) {
            observedShards.add(update.getShardIndex());
        }

        public ImmutableSet<ShardIndex> shards() {
            return observedShards.build();
        }
    }
}

@@ -1707,60 +1707,266 @@
 til.
-Map;%0A%0Aimport static io.spine.base.Identifier.newUuid
+ArrayList;%0Aimport java.util.HashSet;%0Aimport java.util.List;%0Aimport java.util.Map;%0Aimport java.util.Set;%0A%0Aimport static io.spine.base.Identifier.newUuid;%0Aimport static java.util.Collections.synchronizedList;%0Aimport static java.util.Collections.synchronizedSet
 ;%0A%0A/
@@ -6257,29 +6257,12 @@
 nal 
-ImmutableList.Builder
+List
 %3CInb
@@ -6290,30 +6290,42 @@
 s = 
-ImmutableList.builder(
+synchronizedList(new ArrayList%3C%3E()
 );%0A%0A
@@ -6518,16 +6518,37 @@
  return 
+ImmutableList.copyOf(
 rawMessa
@@ -6550,23 +6550,16 @@
 Messages
-.build(
 );%0A     
@@ -6780,28 +6780,11 @@
 nal 
-ImmutableSet.Builder
+Set
 %3CSha
@@ -6813,29 +6813,39 @@
 s = 
-ImmutableSet.builder(
+synchronizedSet(new HashSet%3C%3E()
 );%0A%0A
@@ -7052,16 +7052,36 @@
  return 
+ImmutableSet.copyOf(
 observed
@@ -7090,15 +7090,8 @@
 ards
-.build(
 );%0A 

Make test observers thread-safe.