/*
 * 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 com.google.protobuf.Any;
import io.grpc.stub.StreamObserver;
import io.spine.base.Identifier;
import io.spine.server.integration.ChannelId;
import io.spine.server.integration.ExternalMessage;
import io.spine.server.transport.memory.given.SingleThreadInMemSubscriberTestEnv.ThrowingObserver;
import io.spine.testing.logging.MuteLogging;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;

import static io.spine.base.Identifier.newUuid;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;

@DisplayName("SingleThreadInMemSubscriber should")
class SingleThreadInMemSubscriberTest {

    @SuppressWarnings("unchecked") // OK for testing mocks.
    @Test
    @DisplayName("not halt after observer throws an error")
    @MuteLogging
    void recoverFromObserverError() {
        SingleThreadInMemSubscriber subscriber =
                new SingleThreadInMemSubscriber(ChannelId.getDefaultInstance());

        StreamObserver<ExternalMessage> throwingObserver = new ThrowingObserver();
        subscriber.addObserver(throwingObserver);

        Any id = Identifier.pack(newUuid());
        ExternalMessage externalMessage = ExternalMessage
                .newBuilder()
                .setId(id)
                .build();
        subscriber.onMessage(externalMessage);

        subscriber.removeObserver(throwingObserver);

        StreamObserver observerMock = mock(StreamObserver.class);
        subscriber.addObserver(observerMock);
        subscriber.onMessage(externalMessage);
        verify(observerMock).onNext(externalMessage);
    }
}

@@ -1558,16 +1558,98 @@
 o.mock;%0A
+import static org.mockito.Mockito.spy;%0Aimport static org.mockito.Mockito.timeout;%0A
 import s
@@ -1682,16 +1682,16 @@
 verify;%0A
-
 %0A@Displa
@@ -1775,16 +1775,142 @@
 Test %7B%0A%0A
+    /**%0A     * Wait time to make sure the observer is called.%0A     */%0A    private static final int SYNC_TIMEOUT_MILLS = 500;%0A%0A
     @Sup
@@ -2276,16 +2276,20 @@
 erver = 
+spy(
 new Thro
@@ -2302,16 +2302,17 @@
 server()
+)
 ;%0A      
@@ -2588,16 +2588,103 @@
 essage);
+%0A        verify(throwingObserver, timeout(SYNC_TIMEOUT_MILLS)).onNext(externalMessage);
 %0A%0A      
@@ -2882,32 +2882,32 @@
 ternalMessage);%0A
-
         verify(o
@@ -2917,16 +2917,45 @@
 rverMock
+, timeout(SYNC_TIMEOUT_MILLS)
 ).onNext

Add timeouts to in-mem subscriber tests