Delivery improvements

server/src/main/java/io/spine/server/aggregate/AggregateCommandEndpoint.java

@@ -45,9 +45,9 @@ final class AggregateCommandEndpoint<I, A extends Aggregate<I, ?, ?>>
     }
 
     @Override
-    protected DispatchOutcome invokeDispatcher(A aggregate, CommandEnvelope envelope) {
+    protected DispatchOutcome invokeDispatcher(A aggregate) {
         EntityLifecycle lifecycle = repository().lifecycleOf(aggregate.id());
-        DispatchCommand<I> dispatch = operationFor(lifecycle, aggregate, envelope);
+        DispatchCommand<I> dispatch = operationFor(lifecycle, aggregate, envelope());
         return dispatch.perform();
     }
 
@@ -64,7 +64,8 @@ protected void afterDispatched(I entityId) {
      * @throws IllegalStateException always
      */
     @Override
-    protected void onEmptyResult(A aggregate, CommandEnvelope cmd) throws IllegalStateException {
+    protected void onEmptyResult(A aggregate) throws IllegalStateException {
+        CommandEnvelope cmd = envelope();
         String entityId = aggregate.idAsString();
         String entityClass = aggregate.getClass()
                                       .getName();

server/src/main/java/io/spine/server/aggregate/AggregateEndpoint.java

@@ -92,7 +92,7 @@ private void storeAndPost(A aggregate, DispatchOutcome outcome) {
         } else if (success.hasRejection()) {
             post(success.getRejection());
         } else {
-            onEmptyResult(aggregate, envelope());
+            onEmptyResult(aggregate);
         }
         afterDispatched(aggregate.id());
     }
@@ -118,7 +118,7 @@ private A loadOrCreate(I aggregateId) {
 
     @CanIgnoreReturnValue
     final DispatchOutcome handleAndApplyEvents(A aggregate) {
-        DispatchOutcome outcome = invokeDispatcher(aggregate, envelope());
+        DispatchOutcome outcome = invokeDispatcher(aggregate);
         Success successfulOutcome = outcome.getSuccess();
         return successfulOutcome.hasProducedEvents()
                ? applyProducedEvents(aggregate, outcome)

server/src/main/java/io/spine/server/aggregate/AggregateEventReactionEndpoint.java

@@ -37,8 +37,8 @@ final class AggregateEventReactionEndpoint<I, A extends Aggregate<I, ?, ?>>
     }
 
     @Override
-    protected DispatchOutcome invokeDispatcher(A aggregate, EventEnvelope event) {
-        return aggregate.reactOn(event);
+    protected DispatchOutcome invokeDispatcher(A aggregate) {
+        return aggregate.reactOn(envelope());
     }
 
     @Override
@@ -52,7 +52,7 @@ protected void afterDispatched(I entityId) {
      * updated upon reacting on an event.
      */
     @Override
-    protected void onEmptyResult(A aggregate, EventEnvelope event) {
+    protected void onEmptyResult(A aggregate) {
         // Do nothing.
     }
 }

server/src/main/java/io/spine/server/aggregate/AggregateRepository.java

@@ -20,6 +20,7 @@
 
 package io.spine.server.aggregate;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets.SetView;
 import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
@@ -33,11 +34,13 @@
 import io.spine.server.ServerEnvironment;
 import io.spine.server.aggregate.model.AggregateClass;
 import io.spine.server.commandbus.CommandDispatcher;
+import io.spine.server.delivery.BatchDeliveryListener;
 import io.spine.server.delivery.Delivery;
 import io.spine.server.delivery.Inbox;
 import io.spine.server.delivery.InboxLabel;
 import io.spine.server.entity.EntityLifecycle;
 import io.spine.server.entity.Repository;
+import io.spine.server.entity.RepositoryCache;
 import io.spine.server.event.EventBus;
 import io.spine.server.event.EventDispatcherDelegate;
 import io.spine.server.event.RejectionEnvelope;
@@ -104,6 +107,8 @@ public abstract class AggregateRepository<I, A extends Aggregate<I, ?, ?>>
      */
     private @MonotonicNonNull Inbox<I> inbox;
 
+    private @MonotonicNonNull RepositoryCache<I, A> cache;
+
     /** The number of events to store between snapshots. */
     private int snapshotTrigger = DEFAULT_SNAPSHOT_TRIGGER;
 
@@ -147,10 +152,15 @@ public void registerWith(BoundedContext context) {
             context.importBus()
                    .register(EventImportDispatcher.of(this));
         }
+        initCache(context.isMultitenant());
         initInbox();
         initMirror();
     }
 
+    private void initCache(boolean multitenant) {
+        cache = new RepositoryCache<>(multitenant, this::doLoadOrCreate, this::doStore);
+    }
+
     /**
      * Initializes the {@code Inbox}.
      */
@@ -159,6 +169,17 @@ private void initInbox() {
                                              .delivery();
         inbox = delivery
                 .<I>newInbox(entityStateType())
+                .withBatchListener(new BatchDeliveryListener<I>() {
+                    @Override
+                    public void onStart(I id) {
+                        cache.startCaching(id);
+                    }
+
+                    @Override
+                    public void onEnd(I id) {
+                        cache.stopCaching(id);
+                    }
+                })
                 .addEventEndpoint(InboxLabel.REACT_UPON_EVENT,
                                   e -> new AggregateEventReactionEndpoint<>(this, e))
                 .addEventEndpoint(InboxLabel.IMPORT_EVENT,
@@ -264,6 +285,11 @@ protected AggregateClass<A> toModelClass(Class<A> cls) {
      */
     @Override
     protected final void store(A aggregate) {
+        cache.store(aggregate);
+    }
+
+    @VisibleForTesting
+    protected void doStore(A aggregate) {
         Write<I> operation = Write.operationFor(this, aggregate);
         operation.perform();
     }
@@ -533,6 +559,11 @@ protected AggregateStorage<I> aggregateStorage() {
      * @return loaded or created aggregate instance
      */
     final A loadOrCreate(I id) {
+        return cache.load(id);
+    }
+
+    @VisibleForTesting
+    protected A doLoadOrCreate(I id) {
         A result = load(id).orElseGet(() -> createNew(id));
         return result;
     }

server/src/main/java/io/spine/server/aggregate/EventImportEndpoint.java

@@ -52,8 +52,8 @@ final class EventImportEndpoint<I, A extends Aggregate<I, ?, ?>>
      *         applied}.
      */
     @Override
-    protected DispatchOutcome invokeDispatcher(A aggregate, EventEnvelope eventEnvelope) {
-        Event event = eventEnvelope.outerObject();
+    protected DispatchOutcome invokeDispatcher(A aggregate) {
+        Event event = envelope().outerObject();
         Success.Builder success = Success.newBuilder();
         success.getProducedEventsBuilder()
                .addEvent(event);
@@ -72,8 +72,8 @@ protected void afterDispatched(I entityId) {
     }
 
     @Override
-    protected void onEmptyResult(A aggregate, EventEnvelope event) {
+    protected void onEmptyResult(A aggregate) {
         _error().log("The aggregate `%s` was not modified during the import of the event `%s`.",
-                     aggregate, event);
+                     aggregate, envelope());
     }
 }

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

@@ -0,0 +1,49 @@
+/*
+ * 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.delivery;
+
+/**
+ * Listens to start and end of the {@code Inbox} batch delivery.
+ *
+ * <p>When the consequent messages in the {@code Inbox} are targeting the same entity,
+ * the delivery may be optimized by using either the same transaction or caching the storage
+ * operations while the batch is delivered.
+ *
+ * <p>The implementing classes may define their own behavior and react upon such use cases.
+ */
+public interface BatchDeliveryListener<I> {
+
+    /**
+     * Invoked before the batch delivery to the target with the given ID is started.
+     *
+     * @param id
+     *         the ID of the delivery target
+     */
+    void onStart(I id);
+
+    /**
+     * Invoked after the batch delivery to the target with the given ID is ended.
+     *
+     * @param id
+     *         the ID of the delivery target
+     */
+    void onEnd(I id);
+}