Projection Catch-up

pom.xml

@@ -12,7 +12,7 @@ all modules and does not describe the project structure per-subproject.
 
 <groupId>io.spine</groupId>
 <artifactId>spine-core-java</artifactId>
-<version>1.4.4</version>
+<version>1.4.5</version>
 
 <inceptionYear>2015</inceptionYear>
 

server/src/main/java/io/spine/server/BoundedContext.java

@@ -50,6 +50,7 @@
 import io.spine.system.server.SystemContext;
 import io.spine.system.server.SystemReadSide;
 import io.spine.type.TypeName;
+import io.spine.type.TypeUrl;
 
 import java.util.Optional;
 import java.util.Set;
@@ -357,6 +358,12 @@ public Optional<Repository> findRepository(Class<? extends EntityState> stateCla
         return repository;
     }
 
+    @Internal
+    public Optional<Repository> findRepository(TypeUrl stateType) {
+        Optional<Repository> repository = guard.repositoryFor(stateType);
+        return repository;
+    }
+
     /**
      * Verifies if this Bounded Context contains entities of the passed class.
      *

server/src/main/java/io/spine/server/BoundedContextBuilder.java

@@ -542,6 +542,9 @@ public BoundedContext build() {
 
         registerRepositories(result);
         registerDispatchers(result);
+        ServerEnvironment.instance()
+                         .delivery()
+                         .registerDispatchersIn(result);
         return result;
     }
 

server/src/main/java/io/spine/server/VisibilityGuard.java

@@ -27,6 +27,7 @@
 import io.spine.server.entity.Repository;
 import io.spine.server.entity.model.EntityClass;
 import io.spine.type.TypeName;
+import io.spine.type.TypeUrl;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -107,6 +108,12 @@ Optional<Repository> repositoryFor(Class<? extends EntityState> stateClass) {
         return repositoryAccess.get();
     }
 
+    Optional<Repository> repositoryFor(TypeUrl stateType) {
+        checkNotNull(stateType);
+        RepositoryAccess repositoryAccess = findOrThrow(stateType);
+        return repositoryAccess.get();
+    }
+
     private RepositoryAccess findOrThrow(Class<? extends EntityState> stateClass) {
         RepositoryAccess repository = repositories.get(stateClass);
         if (repository == null) {
@@ -118,6 +125,22 @@ private RepositoryAccess findOrThrow(Class<? extends EntityState> stateClass) {
         return repository;
     }
 
+    private RepositoryAccess findOrThrow(TypeUrl stateType) {
+        Optional<RepositoryAccess> result =
+                repositories.values()
+                            .stream()
+                            .filter((r) -> r.repository.entityStateType()
+                                                       .equals(stateType))
+                            .findFirst();
+        if (!result.isPresent()) {
+            throw newIllegalStateException(
+                    "A repository for the state type `%s` is not registered.",
+                    stateType
+            );
+        }
+        return result.get();
+    }
+
     /**
      * Obtains a set of entity type names by their visibility.
      */

server/src/main/java/io/spine/server/catchup/package-info.java

@@ -0,0 +1,30 @@
+/*
+ * 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.
+ */
+
+/**
+ * A package defining the catch-up related commands and events.
+ */
+@ParametersAreNonnullByDefault
+@CheckReturnValue
+package io.spine.server.catchup;
+
+import com.google.errorprone.annotations.CheckReturnValue;
+
+import javax.annotation.ParametersAreNonnullByDefault;

server/src/main/java/io/spine/server/delivery/CatchUpAlreadyStartedException.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2020, 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.delivery;
+
+import com.google.common.base.Joiner;
+import com.google.common.collect.ImmutableSet;
+import io.spine.type.TypeUrl;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.Set;
+
+/**
+ * An exception telling that the projection catch-up cannot be started, since some of the requested
+ * entities are already catching up.
+ */
+public final class CatchUpAlreadyStartedException extends IllegalStateException {
+
+    private static final long serialVersionUID = 0L;
+
+    private final TypeUrl projectionStateType;
+    private final @Nullable ImmutableSet<Object> requestedIds;
+
+    CatchUpAlreadyStartedException(TypeUrl type, @Nullable Set<?> ids) {
+        super();
+        projectionStateType = type;
+        requestedIds = ids == null ? null : ImmutableSet.copyOf(ids);
+    }
+
+    @Override
+    public String getMessage() {
+        String message = String.format(
+                "Cannot start the catch-up for the `%s` Projection, `%s`. " +
+                        "Another catch-up is already in progress.",
+                projectionStateType, targetsAsString());
+        return message;
+    }
+
+    /**
+     * Returns the type URL of the projection for which the catch-up was requested.
+     */
+    public TypeUrl projectionStateType() {
+        return projectionStateType;
+    }
+
+    /**
+     * Returns the IDs of the targets which were asked to catch up.
+     *
+     * <p>If all the projection entities were specified as a target, returns an empty {@code Set}.
+     */
+    public ImmutableSet<Object> requestedIds() {
+        return requestedIds == null? ImmutableSet.of() : requestedIds;
+    }
+
+    private String targetsAsString() {
+        if (requestedIds == null) {
+            return "[all instances]";
+        }
+        return Joiner.on(',')
+                     .join(requestedIds);
+    }
+}

server/src/main/java/io/spine/server/delivery/CatchUpMessages.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2020, 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.delivery;
+
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.ProtocolStringList;
+import io.spine.server.delivery.event.CatchUpCompleted;
+import io.spine.server.delivery.event.CatchUpStarted;
+import io.spine.server.delivery.event.HistoryEventsRecalled;
+import io.spine.server.delivery.event.HistoryFullyRecalled;
+import io.spine.server.delivery.event.LiveEventsPickedUp;
+import io.spine.server.delivery.event.ShardProcessingRequested;
+import io.spine.server.event.EventFilter;
+import io.spine.server.event.EventStreamQuery;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+/**
+ * A utility class producing the messages related to the {@code CatchUp} process.
+ */
+final class CatchUpMessages {
+
+    private CatchUpMessages() {
+    }
+
+    /**
+     * Creates a {@code CatchUpStarted} event messages with the specified ID.
+     */
+    static CatchUpStarted started(CatchUpId id) {
+        checkNotNull(id);
+        return CatchUpStarted.newBuilder()
+                             .setId(id)
+                             .vBuild();
+    }
+
+    /**
+     * Creates a limit to be used in quering the {@link io.spine.server.event.EventStore
+     * EventStore}.
+     */
+    static EventStreamQuery.Limit limitOf(int value) {
+        return EventStreamQuery.Limit.newBuilder()
+                                     .setValue(value)
+                                     .vBuild();
+    }
+
+    /**
+     * Creates a {@code HistoryEventsRecalled} event messages with the specified ID.
+     */
+    static HistoryEventsRecalled recalled(CatchUpId id) {
+        checkNotNull(id);
+        return HistoryEventsRecalled.newBuilder()
+                                    .setId(id)
+                                    .vBuild();
+    }
+
+    /**
+     * Creates a {@code HistoryFullyRecalled} event messages with the specified ID.
+     */
+    static HistoryFullyRecalled fullyRecalled(CatchUpId id) {
+        checkNotNull(id);
+        return HistoryFullyRecalled.newBuilder()
+                                   .setId(id)
+                                   .vBuild();
+    }
+
+    /**
+     * Wraps the passed list of {@code String}s into a list of {@code EventFilter}s.
+     */
+    static ImmutableList<EventFilter> toFilters(ProtocolStringList rawEventTypes) {
+        checkNotNull(rawEventTypes);
+        return rawEventTypes.stream()
+                            .map(type -> EventFilter
+                                    .newBuilder()
+                                    .setEventType(type)
+                                    .build())
+                            .collect(toImmutableList());
+    }
+
+    /**
+     * Creates a {@code LiveEventsPickedUp} event messages with the specified ID.
+     */
+    static LiveEventsPickedUp liveEventsPickedUp(CatchUpId id) {
+        checkNotNull(id);
+        return LiveEventsPickedUp.newBuilder()
+                                 .setId(id)
+                                 .vBuild();
+    }
+
+    /**
+     * Creates a {@code CatchUpCompleted} event messages with the specified ID.
+     */
+    static CatchUpCompleted catchUpCompleted(CatchUpId id) {
+        checkNotNull(id);
+        return CatchUpCompleted.newBuilder()
+                               .setId(id)
+                               .vBuild();
+    }
+
+    /**
+     * Creates a {@code ShardProcessingRequested} event messages with the specified index.
+     */
+    static ShardProcessingRequested shardProcessingRequested(ShardIndex shardIndex) {
+        checkNotNull(shardIndex);
+        return ShardProcessingRequested
+                .newBuilder()
+                .setId(shardIndex)
+                .vBuild();
+    }
+}

server/src/main/java/io/spine/server/delivery/CatchUpMixin.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2020, 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.delivery;
+
+/**
+ * A mixin for the state of the {@linkplain CatchUp catch-up} {@linkplain CatchUpProcess processes}.
+ */
+public interface CatchUpMixin {
+
+}