Merge pull request #602 from SpineEventEngine/aggregate-storage-events-compacting

Remove an enrichment information from events before storing to `AggregateStorage`

Author: dmytro-grankin

core/src/main/java/io/spine/core/Events.java

@@ -36,6 +36,7 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.spine.protobuf.AnyPacker.unpack;
+import static io.spine.util.Exceptions.newIllegalStateException;
 import static io.spine.validate.Validate.checkNotEmptyOrBlank;
 
 /**
@@ -217,6 +218,56 @@ public static boolean isExternal(EventContext context) {
         return context.getExternal();
     }
 
+    /**
+     * Clears enrichments of the specified event.
+     *
+     * <p>Use this method to decrease a size of an event, if enrichments aren't important.
+     *
+     * <p>A result won't contain:
+     * <ul>
+     *     <li>the enrichment from the event context;</li>
+     *     <li>the enrichment from the first-level origin.</li>
+     * </ul>
+     *
+     * <p>Enrichments will not be removed from second-level and deeper origins,
+     * because it's a heavy performance operation.
+     *
+     * @param event the event to clear enrichments
+     * @return the event without enrichments
+     */
+    @Internal
+    public static Event clearEnrichments(Event event) {
+        final EventContext context = event.getContext();
+        final EventContext.Builder resultContext = context.toBuilder()
+                                                          .clearEnrichment();
+        final EventContext.OriginCase originCase = resultContext.getOriginCase();
+        switch (originCase) {
+            case EVENT_CONTEXT:
+                resultContext.setEventContext(context.getEventContext()
+                                                     .toBuilder()
+                                                     .clearEnrichment());
+                break;
+            case REJECTION_CONTEXT:
+                resultContext.setRejectionContext(context.getRejectionContext()
+                                                         .toBuilder()
+                                                         .clearEnrichment());
+                break;
+            case COMMAND_CONTEXT:
+                // Nothing to remove.
+                break;
+            case ORIGIN_NOT_SET:
+                // Does nothing because there is no origin for this event.
+                break;
+            default:
+                throw newIllegalStateException("Unsupported origin case is encountered: %s",
+                                               originCase);
+        }
+        final Event result = event.toBuilder()
+                                  .setContext(resultContext)
+                                  .build();
+        return result;
+    }
+
     /**
      * The stringifier of event IDs.
      */

core/src/main/proto/spine/core/enrichment.proto

@@ -34,7 +34,7 @@ import "google/protobuf/any.proto";
 // Attributes with additional information (enrichment) for an outer message.
 //
 // A message can be enriched with one or more messages. For example, an `Event` can be enriched with
-// information for easier building of user interface projections. A `Rejection` can be eriched with
+// information for easier building of user interface projections. A `Rejection` can be enriched with
 // additional information on why the command was rejected.
 //
 // If message enrichment should not be performed (e.g. because of performance or security

ext.gradle

@@ -25,7 +25,7 @@
  *  as we want to manage the versions in a single source.
  */
 
-def final SPINE_VERSION = '0.9.77-SNAPSHOT'
+def final SPINE_VERSION = '0.9.78-SNAPSHOT'
 
 ext {
     // The version of the modules in this project.

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

@@ -38,6 +38,7 @@
 import static com.google.common.collect.Lists.newLinkedList;
 import static com.google.protobuf.TextFormat.shortDebugString;
 import static com.google.protobuf.util.Timestamps.checkValid;
+import static io.spine.core.Events.clearEnrichments;
 import static io.spine.util.Exceptions.newIllegalStateException;
 import static io.spine.validate.Validate.checkNotEmptyOrBlank;
 
@@ -156,13 +157,17 @@ public void write(I id, AggregateStateRecord events) {
     /**
      * Writes an event to the storage by an aggregate ID.
      *
+     * <p>Before the storing, {@linkplain io.spine.core.Events#clearEnrichments(Event) enrichments}
+     * will be removed from the event.
+     *
      * @param id    the aggregate ID
      * @param event the event to write
      */
     void writeEvent(I id, Event event) {
         checkNotClosedAndArguments(id, event);
 
-        final AggregateEventRecord record = toStorageRecord(event);
+        final Event eventWithoutEnrichments = clearEnrichments(event);
+        final AggregateEventRecord record = toStorageRecord(eventWithoutEnrichments);
         writeRecord(id, record);
     }
 

server/src/main/java/io/spine/server/event/EEntity.java

@@ -23,7 +23,6 @@
 import com.google.protobuf.Timestamp;
 import io.spine.annotation.Internal;
 import io.spine.core.Event;
-import io.spine.core.EventContext;
 import io.spine.core.EventEnvelope;
 import io.spine.core.EventId;
 import io.spine.core.Events;
@@ -34,10 +33,12 @@
 import javax.annotation.Nullable;
 import java.util.Comparator;
 
-import static io.spine.util.Exceptions.newIllegalStateException;
+import static io.spine.core.Events.clearEnrichments;
 
 /**
- * Stores an event.
+ * An entity for storing an event.
+ *
+ * <p>An underlying event doesn't contain {@linkplain Events#clearEnrichments(Event) enrichments}.
  *
  * @author Alexander Yevsyukov
  * @author Dmytro Dashenkov
@@ -87,8 +88,8 @@ public int compare(EEntity e1, EEntity e2) {
 
     EEntity(Event event) {
         this(event.getId());
-        final Event compactedEvent = compact(event);
-        updateState(compactedEvent);
+        final Event eventWithoutEnrichments = clearEnrichments(event);
+        updateState(eventWithoutEnrichments);
     }
 
     /**
@@ -128,49 +129,4 @@ public String getType() {
         }
         return typeName.value();
     }
-
-    /**
-     * Obtains the compacted version of the event.
-     *
-     * <p>A compacted version doesn't contain:
-     * <ul>
-     *     <li>the enrichment from the event context</li>
-     *     <li>the enrichment from the origin</li>
-     *     <li>nested origins if the origin is {@link EventContext}</li>
-     * </ul>
-     *
-     * @param event the event to compact
-     * @return the compacted event
-     */
-    private static Event compact(Event event) {
-        final EventContext context = event.getContext();
-        final EventContext.Builder resultContext = context.toBuilder()
-                                                          .clearEnrichment();
-        final EventContext.OriginCase originCase = resultContext.getOriginCase();
-        switch (originCase) {
-            case EVENT_CONTEXT:
-                resultContext.setEventContext(context.getEventContext()
-                                                     .toBuilder()
-                                                     .clearOrigin()
-                                                     .clearEnrichment());
-                break;
-            case REJECTION_CONTEXT:
-                resultContext.setRejectionContext(context.getRejectionContext()
-                                                         .toBuilder()
-                                                         .clearEnrichment());
-                break;
-            case COMMAND_CONTEXT:
-                // Does nothing.
-                break;
-            case ORIGIN_NOT_SET:
-                // Does nothing because there is no origin for this event.
-                break;
-            default:
-                throw newIllegalStateException("Unsupported origin case encountered: %s",
-                                               originCase);
-        }
-        return event.toBuilder()
-                    .setContext(resultContext)
-                    .build();
-    }
 }

server/src/test/java/io/spine/server/aggregate/AggregateStorageShould.java

@@ -27,6 +27,8 @@
 import com.google.protobuf.Timestamp;
 import com.google.protobuf.util.Timestamps;
 import io.spine.core.Event;
+import io.spine.core.EventContext;
+import io.spine.core.RejectionContext;
 import io.spine.core.Version;
 import io.spine.server.aggregate.given.Given.StorageRecord;
 import io.spine.server.command.TestEventFactory;
@@ -52,10 +54,12 @@
 import static io.spine.Identifier.newUuid;
 import static io.spine.core.Versions.increment;
 import static io.spine.core.Versions.zero;
+import static io.spine.core.given.GivenEnrichment.withOneAttribute;
 import static io.spine.server.aggregate.given.Given.StorageRecords.sequenceFor;
 import static io.spine.server.command.TestEventFactory.newInstance;
 import static io.spine.time.Durations2.seconds;
 import static io.spine.time.Time.getCurrentTime;
+import static io.spine.validate.Validate.isDefault;
 import static java.lang.Integer.MAX_VALUE;
 import static java.util.Collections.reverse;
 import static org.junit.Assert.assertEquals;
@@ -345,6 +349,65 @@ public void continue_history_reading_if_snapshot_was_not_found_in_first_batch()
         assertEquals(eventCountAfterSnapshot, stateRecord.getEventCount());
     }
 
+    @Test
+    public void not_store_enrichment_for_EventContext() {
+        final EventContext enrichedContext = EventContext.newBuilder()
+                                                         .setEnrichment(withOneAttribute())
+                                                         .build();
+        final Event event = Event.newBuilder()
+                                 .setContext(enrichedContext)
+                                 .setMessage(Any.getDefaultInstance())
+                                 .build();
+        storage.writeEvent(id, event);
+        final EventContext loadedContext = storage.read(newReadRequest(id))
+                                                  .get()
+                                                  .getEvent(0)
+                                                  .getContext();
+        assertTrue(isDefault(loadedContext.getEnrichment()));
+    }
+
+    @Test
+    public void not_store_enrichment_for_origin_of_RejectionContext_type() {
+        final RejectionContext origin = RejectionContext.newBuilder()
+                                                        .setEnrichment(withOneAttribute())
+                                                        .build();
+        final EventContext context = EventContext.newBuilder()
+                                                 .setRejectionContext(origin)
+                                                 .build();
+        final Event event = Event.newBuilder()
+                                 .setContext(context)
+                                 .setMessage(Any.getDefaultInstance())
+                                 .build();
+        storage.writeEvent(id, event);
+        final RejectionContext loadedOrigin = storage.read(newReadRequest(id))
+                                                     .get()
+                                                     .getEvent(0)
+                                                     .getContext()
+                                                     .getRejectionContext();
+        assertTrue(isDefault(loadedOrigin.getEnrichment()));
+    }
+
+    @Test
+    public void not_store_enrichment_for_origin_of_EventContext_type() {
+        final EventContext origin = EventContext.newBuilder()
+                                                .setEnrichment(withOneAttribute())
+                                                .build();
+        final EventContext context = EventContext.newBuilder()
+                                                 .setEventContext(origin)
+                                                 .build();
+        final Event event = Event.newBuilder()
+                                 .setContext(context)
+                                 .setMessage(Any.getDefaultInstance())
+                                 .build();
+        storage.writeEvent(id, event);
+        final EventContext loadedOrigin = storage.read(newReadRequest(id))
+                                                 .get()
+                                                 .getEvent(0)
+                                                 .getContext()
+                                                 .getEventContext();
+        assertTrue(isDefault(loadedOrigin.getEnrichment()));
+    }
+
     @Test(expected = IllegalStateException.class)
     public void throw_exception_if_try_to_write_event_count_to_closed_storage() {
         close(storage);
@@ -402,18 +465,18 @@ protected void writeAll(ProjectId id, Iterable<AggregateEventRecord> records) {
     }
 
     private Iterator<AggregateEventRecord> historyBackward() {
-        final AggregateReadRequest<ProjectId> readRequest = new AggregateReadRequest<>(id, MAX_VALUE);
+        final AggregateReadRequest<ProjectId> readRequest = newReadRequest(id);
         return storage.historyBackward(readRequest);
     }
 
     protected static final Function<AggregateEventRecord, Event> TO_EVENT =
             new Function<AggregateEventRecord, Event>() {
-        @Nullable // return null because an exception won't be propagated in this case
-        @Override
-        public Event apply(@Nullable AggregateEventRecord input) {
-            return (input == null) ? null : input.getEvent();
-        }
-    };
+                @Nullable // return null because an exception won't be propagated in this case
+                @Override
+                public Event apply(@Nullable AggregateEventRecord input) {
+                    return (input == null) ? null : input.getEvent();
+                }
+            };
 
     private static Snapshot newSnapshot(Timestamp time) {
         return Snapshot.newBuilder()