/*
 * Copyright 2017, TeamDev Ltd. 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.integration;

import com.google.common.base.Function;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Multimap;
import com.google.protobuf.Message;
import io.spine.core.BoundedContextName;
import io.spine.protobuf.AnyPacker;
import io.spine.type.TypeUrl;

import java.util.Collection;
import java.util.Set;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * An observer, which reacts to the configuration update messages sent by
 * external entities (such as {@code IntegrationBus}es of other bounded contexts).
 *
 * @author Alex Tymchenko
 */
class ConfigurationChangeObserver extends AbstractChannelObserver implements AutoCloseable {

    private final BoundedContextName boundedContextName;
    private final Function<Class<? extends Message>, BusAdapter<?, ?>> adapterByClass;

    /**
     * Current set of message type URLs, requested by other parties via sending the
     * {@linkplain RequestForExternalMessages configuration messages}, mapped to IDs of their origin
     * bounded contexts.
     */
    private final Multimap<ExternalMessageType, BoundedContextName> requestedTypes =
            HashMultimap.create();

    ConfigurationChangeObserver(BoundedContextName boundedContextName,
                                Function<Class<? extends Message>, BusAdapter<?, ?>> adapterByCls) {
        super(boundedContextName, RequestForExternalMessages.class);
        this.boundedContextName = boundedContextName;
        this.adapterByClass = adapterByCls;
    }

    @Override
    public void handle(ExternalMessage value) {
        final RequestForExternalMessages request = AnyPacker.unpack(value.getOriginalMessage());

        final BoundedContextName originBoundedContextName = value.getBoundedContextName();
        addNewSubscriptions(request.getRequestedMessageTypesList(), originBoundedContextName);
        clearStaleSubscriptions(request.getRequestedMessageTypesList(), originBoundedContextName);
    }

    private void addNewSubscriptions(Iterable<ExternalMessageType> types,
                                     BoundedContextName origin) {
        for (ExternalMessageType newType : types) {
            final Collection<BoundedContextName> contextsWithSameRequest =
                    requestedTypes.get(newType);
            if (contextsWithSameRequest.isEmpty()) {

                // This item has not been requested by anyone yet.
                // Let's create a subscription.
                registerInAdapter(newType);
            }

            requestedTypes.put(newType, origin);
        }
    }

    private void registerInAdapter(ExternalMessageType newType) {
        final Class<Message> wrapperCls = asClassOfMsg(newType.getWrapperTypeUrl());
        final Class<Message> messageCls = asClassOfMsg(newType.getMessageTypeUrl());
        final BusAdapter<?, ?> adapter = getAdapter(wrapperCls);
        adapter.register(messageCls);
    }

    private BusAdapter<?, ?> getAdapter(Class<Message> javaClass) {
        final BusAdapter<?, ?> adapter = adapterByClass.apply(javaClass);
        return checkNotNull(adapter);
    }

    private void clearStaleSubscriptions(Collection<ExternalMessageType> types,
                                         BoundedContextName originBoundedContextName) {

        final Set<ExternalMessageType> toRemove = findStale(types, originBoundedContextName);

        for (ExternalMessageType itemForRemoval : toRemove) {
            final boolean wereNonEmpty = !requestedTypes.get(itemForRemoval)
                                                        .isEmpty();
            requestedTypes.remove(itemForRemoval, originBoundedContextName);
            final boolean emptyNow = requestedTypes.get(itemForRemoval)
                                                   .isEmpty();

            if (wereNonEmpty && emptyNow) {
                unregisterInAdapter(itemForRemoval);
            }
        }
    }

    private void unregisterInAdapter(ExternalMessageType itemForRemoval) {
        // It's now the time to remove the local bus subscription.
        final Class<Message> wrapperCls = asClassOfMsg(itemForRemoval.getWrapperTypeUrl());
        final Class<Message> messageCls = asClassOfMsg(itemForRemoval.getMessageTypeUrl());
        final BusAdapter<?, ?> adapter = getAdapter(wrapperCls);
        adapter.unregister(messageCls);
    }

    private Set<ExternalMessageType> findStale(Collection<ExternalMessageType> types,
                                               BoundedContextName originBoundedContextName) {
        final ImmutableSet.Builder<ExternalMessageType> result = ImmutableSet.builder();

        for (ExternalMessageType previouslyRequestedType : requestedTypes.keySet()) {
            final Collection<BoundedContextName> contextsThatRequested =
                    requestedTypes.get(previouslyRequestedType);

            if (contextsThatRequested.contains(originBoundedContextName) &&
                    !types.contains(previouslyRequestedType)) {

                // The `previouslyRequestedType` item is no longer requested
                // by the bounded context with `originBoundedContextName` name.

                result.add(previouslyRequestedType);
            }
        }
        return result.build();
    }

    @Override
    public String toString() {
        return "Integration bus observer of `RequestedMessageTypes`. " +
                "Bounded Context name = " + boundedContextName.getValue();
    }

    private static Class<Message> asClassOfMsg(String classStr) {
        final TypeUrl typeUrl = TypeUrl.parse(classStr);
        return typeUrl.getJavaClass();
    }

    /**
     * Removes all the current subscriptions from the local buses.
     */
    @Override
    public void close() throws Exception {
        for (ExternalMessageType currentlyRequestedMessage : requestedTypes.keySet()) {
            unregisterInAdapter(currentlyRequestedMessage);
        }
        requestedTypes.clear();
    }
}

@@ -2811,34 +2811,16 @@
 e origin
-BoundedContextName
  = value
@@ -2919,34 +2919,16 @@
 , origin
-BoundedContextName
 );%0A     
@@ -3000,34 +3000,16 @@
 , origin
-BoundedContextName
 );%0A    %7D
@@ -4299,34 +4299,16 @@
 e origin
-BoundedContextName
 ) %7B%0A%0A   
@@ -4377,34 +4377,16 @@
 , origin
-BoundedContextName
 );%0A%0A    
@@ -4644,34 +4644,16 @@
 , origin
-BoundedContextName
 );%0A     
@@ -5507,34 +5507,16 @@
 e origin
-BoundedContextName
 ) %7B%0A    
@@ -5879,34 +5879,16 @@
 s(origin
-BoundedContextName
 ) &&%0A   
@@ -6080,34 +6080,16 @@
  %60origin
-BoundedContextName
 %60 name.%0A

Use just "origin" instead of "originBoundedContextName".