/*
 * 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.transport.memory;

import io.spine.server.integration.ChannelId;
import io.spine.server.integration.ExternalMessage;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * The in-memory subscriber, which uses single-thread delivery of messages.
 *
 * <p>The messages posted by any number of concurrent publishers are delivered to the observers
 * of this subscriber one-by-one in a single thread.
 *
 * <p>This implementation should not be used in production environments, as it is not designed
 * to operate with external transport.
 *
 * @author Alex Tymchenko
 */
class SingleThreadInMemSubscriber extends InMemorySubscriber {

    private final ExecutorService executor = Executors.newSingleThreadExecutor();

    SingleThreadInMemSubscriber(ChannelId channelId) {
        super(channelId);
    }

    @Override
    public void onMessage(ExternalMessage message) {
        executor.submit(new SubmissionJob(message));
    }

    private final class SubmissionJob implements Runnable {

        private final ExternalMessage message;

        private SubmissionJob(ExternalMessage message) {
            this.message = message;
        }

        @Override
        public void run() {
            callObservers(message);
        }
    }

    @Override
    public void close() {
        executor.shutdown();
    }
}

@@ -1048,16 +1048,82 @@
 emory;%0A%0A
+import io.spine.base.Identifier;%0Aimport io.spine.logging.Logging;%0A
 import i
@@ -1212,16 +1212,123 @@
 essage;%0A
+import org.checkerframework.checker.nullness.qual.Nullable;%0A%0Aimport java.util.concurrent.CompletableFuture;
 %0Aimport 
@@ -1403,16 +1403,52 @@
 ecutors;
+%0Aimport java.util.function.Function;
 %0A%0A/**%0A *
@@ -1814,37 +1814,8 @@
 rt.%0A
- *%0A * @author Alex Tymchenko%0A
  */%0A
@@ -1875,16 +1875,35 @@
 scriber 
+implements Logging 
 %7B%0A%0A    p
@@ -2077,207 +2077,480 @@
    @
-Override%0A    public void onMessage(ExternalMessage message) %7B%0A        executor.submit(new SubmissionJob(message));%0A    %7D%0A%0A    private final class SubmissionJob implements Runnable %7B%0A
+SuppressWarnings(%22FutureReturnValueIgnored%22) // Error handling is done manually.%0A    @Override%0A    public void onMessage(ExternalMessage message) %7B%0A        CompletableFuture.runAsync(() -%3E callObservers(message), executor)%0A                         .exceptionally(throwable -%3E logError(throwable, message));%0A    %7D%0A%0A    /**%0A     * Conveniently logs an error in %7B@link CompletableFuture#exceptionally(Function)%7D.%0A     *%0A     * @return %7B@code null%7D always
 %0A     
+*/%0A
    
+ 
 private 
 fina
@@ -2545,21 +2545,60 @@
 private 
-final
+@Nullable Void logError(Throwable throwable,
  Externa
@@ -2613,18 +2613,19 @@
  message
-;%0A
+) %7B
 %0A       
@@ -2629,92 +2629,168 @@
     
-private SubmissionJob(ExternalMessage message) %7B%0A            this.message = message;
+Object id = Identifier.unpack(message.getId());%0A        _error(throwable,%0A               %22Error dispatching an external message %7B%7D with ID %7B%7D to observers: %7B%7D%22,
 %0A   
@@ -2798,90 +2798,123 @@
     
-%7D%0A%0A
        
- @Override%0A        public void run() %7B%0A            callObservers(m
+message.getOriginalMessage().getTypeUrl(),%0A               id,%0A               throwable.getLocalizedM
 essage
+()
 );%0A 
@@ -2920,17 +2920,28 @@
         
-%7D
+return null;
 %0A    %7D%0A%0A

Fix executor errors ignored in `SingleThreadInMemSubscriber`