/*
 * Copyright 2018, 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.integration;

import com.google.common.base.MoreObjects;
import com.google.common.collect.ImmutableSet;
import io.spine.core.BoundedContextName;
import io.spine.core.Event;
import io.spine.core.EventClass;
import io.spine.core.EventEnvelope;
import io.spine.protobuf.AnyPacker;
import io.spine.server.event.AbstractEventSubscriber;
import io.spine.server.transport.Publisher;
import io.spine.server.transport.PublisherHub;

import java.util.Objects;
import java.util.Set;

import static io.spine.server.integration.IntegrationChannels.toId;

/**
 * A subscriber to local {@code EventBus}, which publishes each matching domestic event to
 * a remote channel.
 *
 * <p>The events to subscribe are those that are required by external application components
 * at this moment; their set is determined by the {@linkplain RequestForExternalMessages
 * configuration messages}, received by this instance of {@code IntegrationBus}.
 *
 * @author Alex Tymchenko
 */
final class DomesticEventPublisher extends AbstractEventSubscriber {

    private final BoundedContextName boundedContextName;
    private final PublisherHub publisherHub;
    private final Set<EventClass> eventClasses;

    DomesticEventPublisher(BoundedContextName boundedContextName,
                           PublisherHub publisherHub,
                           EventClass messageClass) {
        super();
        this.boundedContextName = boundedContextName;
        this.publisherHub = publisherHub;
        this.eventClasses = ImmutableSet.of(messageClass);
    }

    @SuppressWarnings("ReturnOfCollectionOrArrayField")     // Returning an immutable impl.
    @Override
    public Set<EventClass> getMessageClasses() {
        return eventClasses;
    }

    @Override
    public Set<String> dispatch(EventEnvelope envelope) {
        Event event = envelope.getOuterObject();
        ExternalMessage msg = ExternalMessages.of(event, boundedContextName);
        ExternalMessageClass messageClass = ExternalMessageClass.of(envelope.getMessageClass());
        ChannelId channelId = toId(messageClass);
        Publisher channel = publisherHub.get(channelId);
        channel.publish(AnyPacker.pack(envelope.getId()), msg);

        return ImmutableSet.of(channel.toString());
    }

    @SuppressWarnings("DuplicateStringLiteralInspection")
    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                          .add("boundedContextName", boundedContextName)
                          .add("eventClasses", eventClasses)
                          .toString();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) {
            return true;
        }
        if (o == null || getClass() != o.getClass()) {
            return false;
        }
        DomesticEventPublisher that = (DomesticEventPublisher) o;
        return Objects.equals(boundedContextName, that.boundedContextName) &&
                Objects.equals(eventClasses, that.eventClasses);
    }

    @Override
    public int hashCode() {
        return Objects.hash(boundedContextName, eventClasses);
    }
}

@@ -1271,16 +1271,49 @@
 velope;%0A
+import io.spine.logging.Logging;%0A
 import i
@@ -1369,37 +1369,29 @@
 r.event.
-AbstractEventSubscrib
+EventDispatch
 er;%0Aimpo
@@ -1521,21 +1521,113 @@
 va.util.
-Set;%0A
+Optional;%0Aimport java.util.Set;%0A%0Aimport static com.google.common.base.Preconditions.checkNotNull;
 %0Aimport 
@@ -2142,39 +2142,51 @@
 her 
-extends AbstractEventSubscriber
+implements EventDispatcher%3CString%3E, Logging
  %7B%0A%0A
@@ -2214,39 +2214,38 @@
 ndedContextName 
-bounded
+origin
 ContextName;%0A   
@@ -2381,23 +2381,22 @@
 extName 
-bounded
+origin
 ContextN
@@ -2538,23 +2538,22 @@
    this.
-bounded
+origin
 ContextN
@@ -2558,23 +2558,22 @@
 tName = 
-bounded
+origin
 ContextN
@@ -3054,23 +3054,22 @@
 (event, 
-bounded
+origin
 ContextN
@@ -3394,32 +3394,642 @@
 ring());%0A    %7D%0A%0A
+    @Override%0A    public void onError(EventEnvelope envelope, RuntimeException exception) %7B%0A        checkNotNull(envelope);%0A        checkNotNull(envelope);%0A        _error(exception,%0A               %22Error publishing event (class: %60%25s%60, ID: %60%25s%60) from bounded context %60%25s%60.%22,%0A               envelope.getMessageClass(), envelope.getId(), originContextName);%0A    %7D%0A%0A    @Override%0A    public Set%3CEventClass%3E getExternalEventClasses() %7B%0A        return ImmutableSet.of();%0A    %7D%0A%0A    @Override%0A    public Optional%3CExternalMessageDispatcher%3CString%3E%3E createExternalDispatcher() %7B%0A        return Optional.empty();%0A    %7D%0A%0A
     @SuppressWar
@@ -4195,23 +4195,22 @@
   .add(%22
-bounded
+origin
 ContextN
@@ -4215,23 +4215,22 @@
 tName%22, 
-bounded
+origin
 ContextN
@@ -4640,23 +4640,22 @@
 .equals(
-bounded
+origin
 ContextN
@@ -4664,23 +4664,22 @@
 e, that.
-bounded
+origin
 ContextN
@@ -4828,23 +4828,22 @@
 ts.hash(
-bounded
+origin
 ContextN

Make `DomesticEventPublisher` implement `EventDispatcher` directly.