<commit_before>/*
 * 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.event;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Optional;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.Message;
import io.grpc.stub.StreamObserver;
import io.spine.annotation.Internal;
import io.spine.annotation.Subscribe;
import io.spine.base.Event;
import io.spine.base.EventContext;
import io.spine.base.Response;
import io.spine.envelope.EventEnvelope;
import io.spine.io.StreamObservers;
import io.spine.server.event.enrich.EventEnricher;
import io.spine.server.outbus.CommandOutputBus;
import io.spine.server.outbus.OutputDispatcherRegistry;
import io.spine.server.storage.StorageFactory;
import io.spine.type.EventClass;
import io.spine.validate.ConstraintViolation;
import io.spine.validate.MessageValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.spine.server.transport.Statuses.invalidArgumentWithCause;

/**
 * Dispatches incoming events to subscribers, and provides ways for registering those subscribers.
 *
 * <h2>Receiving Events</h2>
 * <p>To receive event messages a subscriber object should:
 * <ol>
 * <li>Expose a {@code public} method that accepts an event message as the first parameter
 * and an {@link EventContext EventContext} as the second
 * (optional) parameter.
 * <li>Mark the method with the {@link Subscribe @Subscribe} annotation.
 * <li>{@linkplain #register(io.spine.server.bus.MessageDispatcher)} Register} with an
 * instance of {@code EventBus} directly, or rely on message delivery
 * from an {@link EventDispatcher}. An example of such a dispatcher is
 * {@link io.spine.server.projection.ProjectionRepository ProjectionRepository}
 * </ol>
 *
 * <p><strong>Note:</strong> A subscriber method cannot accept just {@link Message} as
 * the first parameter. It must be an <strong>exact type</strong> of the event message
 * that needs to be handled.
 *
 * <h2>Posting Events</h2>
 * <p>Events are posted to an EventBus using {@link #post(Message, StreamObserver)} method.
 * Normally this is done by an
 * {@linkplain io.spine.server.aggregate.AggregateRepository AggregateRepository} in the process
 * of handling a command,
 * or by a {@linkplain io.spine.server.procman.ProcessManager ProcessManager}.
 *
 * <p>The passed {@link Event} is stored in the {@link EventStore} associated with
 * the {@code EventBus} <strong>before</strong> it is passed to subscribers.
 *
 * <p>The delivery of the events to the subscribers and dispatchers is performed by
 * the {@link DispatcherEventDelivery} strategy associated with
 * this instance of the {@code EventBus}.
 *
 * <p>If there is no subscribers or dispatchers for the posted event, the fact is
 * logged as warning, with no further processing.
 *
 * @author <NAME>
 * @author <NAME>
 * @author <NAME>
 * @see io.spine.server.projection.Projection Projection
 * @see Subscribe @Subscribe
 */
public class EventBus extends CommandOutputBus<Event, EventEnvelope, EventClass, EventDispatcher> {

    /*
     * NOTE: Even though, the EventBus has a private constructor and
     * is not supposed to be derived, we do not make this class final
     * in order to be able to spy() on it from Mockito (which cannot
     * spy on final or anonymous classes).
     */

    /** The {@code EventStore} to which put events before they get handled. */
    private final EventStore eventStore;

    /** The validator for events posted to the bus. */
    private final MessageValidator eventValidator;

    /** The enricher for posted events or {@code null} if the enrichment is not supported. */
    @Nullable
    private EventEnricher enricher;

    /**
     * Creates new instance by the passed builder.
     */
    private EventBus(Builder builder) {
        super(checkNotNull(builder.dispatcherEventDelivery));
        this.eventStore = builder.eventStore;
        this.eventValidator = builder.eventValidator;
        this.enricher = builder.enricher;
    }

    /** Creates a builder for new {@code EventBus}. */
    public static Builder newBuilder() {
        return new Builder();
    }

    @VisibleForTesting
    MessageValidator getEventValidator() {
        return eventValidator;
    }

    @VisibleForTesting
    @Nullable
    EventEnricher getEnricher() {
        return enricher;
    }

    @VisibleForTesting
    Set<EventDispatcher> getDispatchers(EventClass eventClass) {
        return registry().getDispatchers(eventClass);
    }

    @VisibleForTesting
    boolean hasDispatchers(EventClass eventClass) {
        return registry().hasDispatchersFor(eventClass);
    }

    @Override
    public void handleDeadMessage(EventEnvelope message) {
        final Event event = message.getOuterObject();
        log().warn("No subscriber or dispatcher defined for the event class: {}",
                   event.getClass()
                        .getName());
    }

    @Override
    protected OutputDispatcherRegistry<EventClass, EventDispatcher> createRegistry() {
        return new EventDispatcherRegistry();
    }

    @Override
    protected EventEnvelope parcel(Event message) {
        return EventEnvelope.of(message);
    }

    /** Returns {@link EventStore} associated with the bus. */
    public EventStore getEventStore() {
        return eventStore;
    }

    /**
     * Validates and posts the event for handling.
     *
     * <p>Does performs the same as the
     * {@linkplain CommandOutputBus#post(Message, StreamObserver)} parent method}, but does not
     * require any response observer.
     *
     * <p>This method should be used if the callee does not care about the event acknowledgement.
     *
     * @param event the event to be handled
     * @see CommandOutputBus#post(Message, StreamObserver)
     */
    public void post(Event event) {
        post(event, StreamObservers.<Response>noOpObserver());
    }

    public void post(Iterable<Event> events) {
        post(events, StreamObservers.<Response>noOpObserver());
    }

    @Override
    protected Event enrich(Event event) {
        if (enricher == null || !enricher.canBeEnriched(event)) {
            return event;
        }
        final Event enriched = enricher.enrich(event);
        return enriched;
    }

    @Override
    protected void store(Event event) {
        eventStore.append(event);
    }
<commit_msg>Yet more Javadoc.<commit_after>/*
 * 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.event;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Optional;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.protobuf.Message;
import io.grpc.stub.StreamObserver;
import io.spine.annotation.Internal;
import io.spine.annotation.Subscribe;
import io.spine.base.Event;
import io.spine.base.EventContext;
import io.spine.base.Response;
import io.spine.envelope.EventEnvelope;
import io.spine.io.StreamObservers;
import io.spine.server.event.enrich.EventEnricher;
import io.spine.server.outbus.CommandOutputBus;
import io.spine.server.outbus.OutputDispatcherRegistry;
import io.spine.server.storage.StorageFactory;
import io.spine.type.EventClass;
import io.spine.validate.ConstraintViolation;
import io.spine.validate.MessageValidator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static io.spine.server.transport.Statuses.invalidArgumentWithCause;

/**
 * Dispatches incoming events to subscribers, and provides ways for registering those subscribers.
 *
 * <h2>Receiving Events</h2>
 * <p>To receive event messages a subscriber object should:
 * <ol>
 * <li>Expose a {@code public} method that accepts an event message as the first parameter
 * and an {@link EventContext EventContext} as the second
 * (optional) parameter.
 * <li>Mark the method with the {@link Subscribe @Subscribe} annotation.
 * <li>{@linkplain #register(io.spine.server.bus.MessageDispatcher)} Register} with an
 * instance of {@code EventBus} directly, or rely on message delivery
 * from an {@link EventDispatcher}. An example of such a dispatcher is
 * {@link io.spine.server.projection.ProjectionRepository ProjectionRepository}
 * </ol>
 *
 * <p><strong>Note:</strong> A subscriber method cannot accept just {@link Message} as
 * the first parameter. It must be an <strong>exact type</strong> of the event message
 * that needs to be handled.
 *
 * <h2>Posting Events</h2>
 * <p>Events are posted to an EventBus using {@link #post(Message, StreamObserver)} method.
 * Normally this is done by an
 * {@linkplain io.spine.server.aggregate.AggregateRepository AggregateRepository} in the process
 * of handling a command,
 * or by a {@linkplain io.spine.server.procman.ProcessManager ProcessManager}.
 *
 * <p>The passed {@link Event} is stored in the {@link EventStore} associated with
 * the {@code EventBus} <strong>before</strong> it is passed to subscribers.
 *
 * <p>The delivery of the events to the subscribers and dispatchers is performed by
 * the {@link DispatcherEventDelivery} strategy associated with
 * this instance of the {@code EventBus}.
 *
 * <p>If there is no subscribers or dispatchers for the posted event, the fact is
 * logged as warning, with no further processing.
 *
 * @author <NAME>
 * @author <NAME>
 * @author <NAME>
 * @see io.spine.server.projection.Projection Projection
 * @see Subscribe @Subscribe
 */
public class EventBus extends CommandOutputBus<Event, EventEnvelope, EventClass, EventDispatcher> {

    /*
     * NOTE: Even though, the EventBus has a private constructor and
     * is not supposed to be derived, we do not make this class final
     * in order to be able to spy() on it from Mockito (which cannot
     * spy on final or anonymous classes).
     */

    /** The {@code EventStore} to which put events before they get handled. */
    private final EventStore eventStore;

    /** The validator for events posted to the bus. */
    private final MessageValidator eventValidator;

    /** The enricher for posted events or {@code null} if the enrichment is not supported. */
    @Nullable
    private EventEnricher enricher;

    /**
     * Creates new instance by the passed builder.
     */
    private EventBus(Builder builder) {
        super(checkNotNull(builder.dispatcherEventDelivery));
        this.eventStore = builder.eventStore;
        this.eventValidator = builder.eventValidator;
        this.enricher = builder.enricher;
    }

    /** Creates a builder for new {@code EventBus}. */
    public static Builder newBuilder() {
        return new Builder();
    }

    @VisibleForTesting
    MessageValidator getEventValidator() {
        return eventValidator;
    }

    @VisibleForTesting
    @Nullable
    EventEnricher getEnricher() {
        return enricher;
    }

    @VisibleForTesting
    Set<EventDispatcher> getDispatchers(EventClass eventClass) {
        return registry().getDispatchers(eventClass);
    }

    @VisibleForTesting
    boolean hasDispatchers(EventClass eventClass) {
        return registry().hasDispatchersFor(eventClass);
    }

    @Override
    public void handleDeadMessage(EventEnvelope message) {
        final Event event = message.getOuterObject();
        log().warn("No subscriber or dispatcher defined for the event class: {}",
                   event.getClass()
                        .getName());
    }

    @Override
    protected OutputDispatcherRegistry<EventClass, EventDispatcher> createRegistry() {
        return new EventDispatcherRegistry();
    }

    @Override
    protected EventEnvelope parcel(Event message) {
        return EventEnvelope.of(message);
    }

    /** Returns {@link EventStore} associated with the bus. */
    public EventStore getEventStore() {
        return eventStore;
    }

    /**
     * Validates and posts the event for handling.
     *
     * <p>Performs the same action as the
     * {@linkplain CommandOutputBus#post(Message, StreamObserver)} parent method}, but does not
     * require any response observer.
     *
     * <p>This method should be used if the callee does not care about the event acknowledgement.
     *
     * @param event the event to be handled
     * @see CommandOutputBus#post(Message, StreamObserver)
     */
    public void post(Event event) {
        post(event, StreamObservers.<Response>noOpObserver());
    }

    /**
     * Validates and posts the event for handling.
     *
     * <p>Performs the same action as the
     * {@linkplain CommandOutputBus#post(Iterable, StreamObserver)} parent method}, but does not
     * require any response observer.
     *
     * <p>This method should be used if the callee does not care about the events acknowledgement.
     *
     * @param events the events to be handled
     * @see CommandOutputBus#post(Message, StreamObserver)
     */
    public void post(Iterable<Event> events) {
        post(events, StreamObservers.<Response>noOpObserver());
    }

    @Override
    protected Event enrich(Event event) {
        if (enricher == null || !enricher.canBeEnriched(event)) {
            return event;
        }
        final Event enriched = enricher.enrich(event);
        return enriched;
    }

    @Override
    protected void store(Event event) {
        eventStore.append(event);
    }

{'fraction_non_alphanumeric': 0.08886638394835115, 'fraction_numerical': 0.0012659029052471675, 'mean_word_length': 3.6415981198589895, 'pattern_counts': {'":': 0, '<': 69, '<?xml version=': 0, '>': 69, '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': '2272271', 'n_tokens_mistral': 4400, 'n_tokens_neox': 4198, 'n_words': 1492}