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

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
import com.google.protobuf.Message;
import com.google.protobuf.Timestamp;
import io.spine.annotation.Internal;
import io.spine.core.Event;
import io.spine.server.entity.EventDispatchingRepository;
import io.spine.server.entity.StorageConverter;
import io.spine.server.event.EventFilter;
import io.spine.server.event.EventStreamQuery;
import io.spine.server.event.model.SubscriberMethod;
import io.spine.server.event.store.EventStore;
import io.spine.server.integration.ExternalMessageClass;
import io.spine.server.integration.ExternalMessageDispatcher;
import io.spine.server.integration.ExternalMessageEnvelope;
import io.spine.server.projection.model.ProjectionClass;
import io.spine.server.route.StateUpdateRouting;
import io.spine.server.stand.Stand;
import io.spine.server.storage.RecordStorage;
import io.spine.server.storage.StorageFactory;
import io.spine.server.type.EventClass;
import io.spine.server.type.EventEnvelope;
import io.spine.type.TypeName;
import org.checkerframework.checker.nullness.qual.Nullable;

import java.util.Optional;
import java.util.Set;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.spine.option.EntityOption.Kind.PROJECTION;
import static io.spine.server.projection.model.ProjectionClass.asProjectionClass;
import static io.spine.server.route.EventRoute.byProducerId;
import static io.spine.util.Exceptions.newIllegalStateException;

/**
 * Abstract base for repositories managing {@link Projection}s.
 *
 * @param <I> the type of IDs of projections
 * @param <P> the type of projections
 * @param <S> the type of projection state messages
 */
public abstract class ProjectionRepository<I, P extends Projection<I, S, ?>, S extends Message>
        extends EventDispatchingRepository<I, P, S> {

    /** An underlying entity storage used to store projections. */
    private RecordStorage<I> recordStorage;

    /**
     * Creates a new {@code ProjectionRepository}.
     */
    protected ProjectionRepository() {
        super(byProducerId());
    }

    /**
     * Initializes the repository during its registration.
     *
     * <p>If projections of this repository are {@linkplain io.spine.core.Subscribe subscribed} to
     * entity state updates, a routing for these updates is {@linkplain #createStateRouting()
     * created}. If one of the states of entities cannot be routed during the created schema,
     * {@code IllegalStateException} will be thrown.
     *
     * @throws IllegalStateException
     *          if the state routing does not cover one of the entity state types to which
     *          the entities are subscribed
     */
    @Override
    @OverridingMethodsMustInvokeSuper
    protected void init() throws IllegalStateException{
        super.init();
        if (projectionClass().subscribesToStates()) {
            eventRouting().routeStateUpdates(createStateRouting());
            validateStateRouting();
        }
    }

    /**
     * Creates {@code StateUpdateRouting} used by the repository.
     *
     * <p>Default implementation provides {@linkplain StateUpdateRouting#newInstance()
     * default instance} of the {@code StateUpdateRouting}. Overriding repository classes
     * may customize the routing if the default schema does not satisfy the routing needs.
     */
    protected StateUpdateRouting<I> createStateRouting() {
        return StateUpdateRouting.newInstance();
    }

    /**
     * Validates routing schema for types of state messages (if any) that this repository dispatches
     * to its entities.
     *
     * @throws IllegalStateException
     *          if a message type cannot be dispatched
     */
    private void validateStateRouting() throws IllegalStateException {
        if (!projectionClass().subscribesToStates()) {
            return;
        }
        //TODO:2019-05-13:alexander.yevsyukov: Implement
    }

    @VisibleForTesting
    static Timestamp nullToDefault(@Nullable Timestamp timestamp) {
        return timestamp == null
               ? Timestamp.getDefaultInstance()
               : timestamp;
    }

    /** Obtains {@link EventStore} from which to get events during catch-up. */
    EventStore eventStore() {
        return context()
                .eventBus()
                .eventStore();
    }

    /** Obtains class information of projection managed by this repository. */
    private ProjectionClass<P> projectionClass() {
        return (ProjectionClass<P>) entityModelClass();
    }

    @Internal
    @Override
    protected final ProjectionClass<P> toModelClass(Class<P> cls) {
        return asProjectionClass(cls);
    }

    @Override
    public P create(I id) {
        P projection = super.create(id);
        lifecycleOf(id).onEntityCreated(PROJECTION);
        return projection;
    }

    /**
     * {@inheritDoc}
     *
     * <p>Ensures there is at least one event subscriber method (external or domestic) declared
     * by the class of the projection. Throws an {@code IllegalStateException} otherwise.
     */
    @Override
    @OverridingMethodsMustInvokeSuper
    public void onRegistered() {
        super.onRegistered();
        ensureDispatchesEvents();
        subscribeToSystemEvents();
    }

    private void ensureDispatchesEvents() {
        boolean noEventSubscriptions = !dispatchesEvents();
        if (noEventSubscriptions) {
            boolean noExternalSubscriptions = !dispatchesExternalEvents();
            if (noExternalSubscriptions) {
                throw newIllegalStateException(
                        "Projections of the repository `%s` have neither domestic nor external " +
                                "event subscriptions.", this);
            }
        }
    }

    private void subscribeToSystemEvents() {
        ProjectionSystemEventWatcher<I> systemSubscriber =
                new ProjectionSystemEventWatcher<>(this);
        systemSubscriber.registerIn(context());
    }

    @Override
    public Optional<ExternalMessageDispatcher<I>> createExternalDispatcher() {
        if (!dispatchesExternalEvents()) {
            return Optional.empty();
        }
        return Optional.of(new ProjectionExternalEventDispatcher());
    }
<commit_msg>Fix diag message layout<commit_after>/*
 * 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.projection;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableSet;
import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
import com.google.protobuf.Message;
import com.google.protobuf.Timestamp;
import io.spine.annotation.Internal;
import io.spine.core.Event;
import io.spine.server.entity.EventDispatchingRepository;
import io.spine.server.entity.StorageConverter;
import io.spine.server.event.EventFilter;
import io.spine.server.event.EventStreamQuery;
import io.spine.server.event.model.SubscriberMethod;
import io.spine.server.event.store.EventStore;
import io.spine.server.integration.ExternalMessageClass;
import io.spine.server.integration.ExternalMessageDispatcher;
import io.spine.server.integration.ExternalMessageEnvelope;
import io.spine.server.projection.model.ProjectionClass;
import io.spine.server.route.StateUpdateRouting;
import io.spine.server.stand.Stand;
import io.spine.server.storage.RecordStorage;
import io.spine.server.storage.StorageFactory;
import io.spine.server.type.EventClass;
import io.spine.server.type.EventEnvelope;
import io.spine.type.TypeName;
import org.checkerframework.checker.nullness.qual.Nullable;

import java.util.Optional;
import java.util.Set;

import static com.google.common.base.Preconditions.checkNotNull;
import static io.spine.option.EntityOption.Kind.PROJECTION;
import static io.spine.server.projection.model.ProjectionClass.asProjectionClass;
import static io.spine.server.route.EventRoute.byProducerId;
import static io.spine.util.Exceptions.newIllegalStateException;

/**
 * Abstract base for repositories managing {@link Projection}s.
 *
 * @param <I> the type of IDs of projections
 * @param <P> the type of projections
 * @param <S> the type of projection state messages
 */
public abstract class ProjectionRepository<I, P extends Projection<I, S, ?>, S extends Message>
        extends EventDispatchingRepository<I, P, S> {

    /** An underlying entity storage used to store projections. */
    private RecordStorage<I> recordStorage;

    /**
     * Creates a new {@code ProjectionRepository}.
     */
    protected ProjectionRepository() {
        super(byProducerId());
    }

    /**
     * Initializes the repository during its registration.
     *
     * <p>If projections of this repository are {@linkplain io.spine.core.Subscribe subscribed} to
     * entity state updates, a routing for these updates is {@linkplain #createStateRouting()
     * created}. If one of the states of entities cannot be routed during the created schema,
     * {@code IllegalStateException} will be thrown.
     *
     * @throws IllegalStateException
     *          if the state routing does not cover one of the entity state types to which
     *          the entities are subscribed
     */
    @Override
    @OverridingMethodsMustInvokeSuper
    protected void init() throws IllegalStateException{
        super.init();
        if (projectionClass().subscribesToStates()) {
            eventRouting().routeStateUpdates(createStateRouting());
            validateStateRouting();
        }
    }

    /**
     * Creates {@code StateUpdateRouting} used by the repository.
     *
     * <p>Default implementation provides {@linkplain StateUpdateRouting#newInstance()
     * default instance} of the {@code StateUpdateRouting}. Overriding repository classes
     * may customize the routing if the default schema does not satisfy the routing needs.
     */
    protected StateUpdateRouting<I> createStateRouting() {
        return StateUpdateRouting.newInstance();
    }

    /**
     * Validates routing schema for types of state messages (if any) that this repository dispatches
     * to its entities.
     *
     * @throws IllegalStateException
     *          if a message type cannot be dispatched
     */
    private void validateStateRouting() throws IllegalStateException {
        if (!projectionClass().subscribesToStates()) {
            return;
        }
        //TODO:2019-05-13:alexander.yevsyukov: Implement
    }

    @VisibleForTesting
    static Timestamp nullToDefault(@Nullable Timestamp timestamp) {
        return timestamp == null
               ? Timestamp.getDefaultInstance()
               : timestamp;
    }

    /** Obtains {@link EventStore} from which to get events during catch-up. */
    EventStore eventStore() {
        return context()
                .eventBus()
                .eventStore();
    }

    /** Obtains class information of projection managed by this repository. */
    private ProjectionClass<P> projectionClass() {
        return (ProjectionClass<P>) entityModelClass();
    }

    @Internal
    @Override
    protected final ProjectionClass<P> toModelClass(Class<P> cls) {
        return asProjectionClass(cls);
    }

    @Override
    public P create(I id) {
        P projection = super.create(id);
        lifecycleOf(id).onEntityCreated(PROJECTION);
        return projection;
    }

    /**
     * {@inheritDoc}
     *
     * <p>Ensures there is at least one event subscriber method (external or domestic) declared
     * by the class of the projection. Throws an {@code IllegalStateException} otherwise.
     */
    @Override
    @OverridingMethodsMustInvokeSuper
    public void onRegistered() {
        super.onRegistered();
        ensureDispatchesEvents();
        subscribeToSystemEvents();
    }

    private void ensureDispatchesEvents() {
        boolean noEventSubscriptions = !dispatchesEvents();
        if (noEventSubscriptions) {
            boolean noExternalSubscriptions = !dispatchesExternalEvents();
            if (noExternalSubscriptions) {
                throw newIllegalStateException(
                        "Projections of the repository `%s` have neither domestic nor external" +
                                " event subscriptions.", this);
            }
        }
    }

    private void subscribeToSystemEvents() {
        ProjectionSystemEventWatcher<I> systemSubscriber =
                new ProjectionSystemEventWatcher<>(this);
        systemSubscriber.registerIn(context());
    }

    @Override
    public Optional<ExternalMessageDispatcher<I>> createExternalDispatcher() {
        if (!dispatchesExternalEvents()) {
            return Optional.empty();
        }
        return Optional.of(new ProjectionExternalEventDispatcher());
    }

{'fraction_non_alphanumeric': 0.0847913862718708, 'fraction_numerical': 0.0016150740242261105, 'mean_word_length': 3.2901270207852193, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1465649', 'n_tokens_mistral': 3983, 'n_tokens_neox': 3805, 'n_words': 1161}