<commit_before>package de.oneos.eventstore.inmemory

import org.junit.*

import static org.mockito.Mockito.mock
import static org.mockito.Mockito.reset
import static org.mockito.Mockito.verify
import static org.mockito.Mockito.verifyNoMoreInteractions

import org.cqrest.reactive.test.MockObserver
import static de.oneos.eventstore.inmemory.AnEventEnvelope.anEventEnvelope

import rx.plugins.RxJavaErrorHandler
import rx.plugins.RxJavaPlugins

import de.oneos.eventsourcing.EventEnvelope
import de.oneos.eventstore.EventCollisionOccurred
import de.oneos.eventstore.EventStore_ContractTest


class InMemoryEventStoreTest extends EventStore_ContractTest {

    private static RxJavaErrorHandler mockErrorHandler

    InMemoryEventStore eventStore

    @BeforeClass
    static void setUpRxJavaErrorHandler() {
        mockErrorHandler = mock(RxJavaErrorHandler)
        // Might lead to interacting tests
        RxJavaPlugins.getInstance().registerErrorHandler(mockErrorHandler)
    }

    @Override
    @Before
    public void setUp() {
        super.setUp()

        eventStore = new InMemoryEventStore()
        reset(mockErrorHandler)
    }

    @Test
    void observe__should_send_EventEnvelopes_to_subscribed_Observers() {
        eventStore.history = [
          anEventEnvelope().withAggregateId(ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
          anEventEnvelope().withAggregateId(ANOTHER_ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
        ]
        final MockObserver observer = new MockObserver(eventStore.history)

        eventStore.observe().subscribe(observer)

        observer.assertReceivedEvents()
    }

    @Test
    void observe__should_filter_EventEnvelopes_by_event_names() {
        final notMatching = [ anEventEnvelope().withAggregateId(ORDER_ID).withEventName('Order line added').build() ]
        final matching =    [ anEventEnvelope().withAggregateId(ANOTHER_ORDER_ID).withEventName('Order line removed').build() ]
        eventStore.history = notMatching + matching
        final MockObserver observer = new MockObserver(matching)

        eventStore.observe(eventName: 'Order line removed').subscribe(observer)

        observer.assertReceivedEvents()
    }


    @Test
    void should_pass_new_persisted_events_to_subscribed_Observers() {
        final notMatching = [ anEventEnvelope().withEventName('Order line added').build() ]
        final matching =    [ anEventEnvelope().withEventName('Order line removed').build() ]

        final MockObserver observer = new MockObserver(matching)
        eventStore.observe(eventName: 'Order line removed').subscribe(observer)

        eventStore.saveEnvelopes(notMatching + matching)<commit_msg>CLOSED #22 Asserted that observed streams are never completed<commit_after>package de.oneos.eventstore.inmemory

import org.junit.*

import static org.mockito.Mockito.mock
import static org.mockito.Mockito.never
import static org.mockito.Mockito.reset
import static org.mockito.Mockito.verify
import static org.mockito.Mockito.verifyNoMoreInteractions

import org.cqrest.reactive.test.MockObserver
import static de.oneos.eventstore.inmemory.AnEventEnvelope.anEventEnvelope

import rx.plugins.RxJavaErrorHandler
import rx.plugins.RxJavaPlugins

import de.oneos.eventsourcing.EventEnvelope
import de.oneos.eventstore.EventCollisionOccurred
import de.oneos.eventstore.EventStore_ContractTest


class InMemoryEventStoreTest extends EventStore_ContractTest {

    private static RxJavaErrorHandler mockErrorHandler

    InMemoryEventStore eventStore

    @BeforeClass
    static void setUpRxJavaErrorHandler() {
        mockErrorHandler = mock(RxJavaErrorHandler)
        // Might lead to interacting tests
        RxJavaPlugins.getInstance().registerErrorHandler(mockErrorHandler)
    }

    @Override
    @Before
    public void setUp() {
        super.setUp()

        eventStore = new InMemoryEventStore()
        reset(mockErrorHandler)
    }

    @Test
    void observe__should_send_EventEnvelopes_to_subscribed_Observers() {
        eventStore.history = [
          anEventEnvelope().withAggregateId(ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
          anEventEnvelope().withAggregateId(ANOTHER_ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
        ]
        final MockObserver observer = new MockObserver(eventStore.history)

        eventStore.observe().subscribe(observer)

        observer.assertReceivedEvents()
    }

    @Test
    void observe__should_not_complete_streams_of_Observers() {
        eventStore.history = [
          anEventEnvelope().withAggregateId(ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
          anEventEnvelope().withAggregateId(ANOTHER_ORDER_ID).withEventName('Order line was added').withEventAttributes(article: null).build(),
        ]
        final org.cqrest.reactive.Observer<EventEnvelope> observer = mock(org.cqrest.reactive.Observer)

        eventStore.observe().subscribe(observer)

        verify(observer, never()).onCompleted()
    }

    @Test
    void observe__should_filter_EventEnvelopes_by_event_names() {
        final notMatching = [ anEventEnvelope().withAggregateId(ORDER_ID).withEventName('Order line added').build() ]
        final matching =    [ anEventEnvelope().withAggregateId(ANOTHER_ORDER_ID).withEventName('Order line removed').build() ]
        eventStore.history = notMatching + matching
        final MockObserver observer = new MockObserver(matching)

        eventStore.observe(eventName: 'Order line removed').subscribe(observer)

        observer.assertReceivedEvents()
    }


    @Test
    void should_pass_new_persisted_events_to_subscribed_Observers() {
        final notMatching = [ anEventEnvelope().withEventName('Order line added').build() ]
        final matching =    [ anEventEnvelope().withEventName('Order line removed').build() ]

        final MockObserver observer = new MockObserver(matching)
        eventStore.observe(eventName: 'Order line removed').subscribe(observer)

        eventStore.saveEnvelopes(notMatching + matching)
{'fraction_non_alphanumeric': 0.08662175168431184, 'fraction_numerical': 0.0003208213025344883, 'mean_word_length': 4.19150707743547, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1896001', 'n_tokens_mistral': 1734, 'n_tokens_neox': 1685, 'n_words': 299}