Merge pull request #119 from SpineEventEngine/update-to-responseformat-api

Update to the latest `Storage` API.

Author: armiol

datastore/config/index.yaml

@@ -10,23 +10,10 @@ indexes:
     - name: type
     - name: created
 
-  - kind: spine.system.server.EntityHistory
+  - kind: spine.server.delivery.InboxMessage
     properties:
-    - name: aggregate_id
-    - name: version
-      direction: desc
-    - name: created
-      direction: desc
-    - name: snapshot
-
-  - kind: spine.system.server.CommandLifecycle
-    properties:
-    - name: aggregate_id
-    - name: version
-      direction: desc
-    - name: created
-      direction: desc
-    - name: snapshot
+      - name: inbox_shard
+      - name: when_received
 
   # Indexes for the Aggregates.
   #

datastore/src/main/java/io/spine/server/storage/datastore/DsLookupByQueries.java

@@ -25,6 +25,7 @@
 import com.google.common.collect.Streams;
 import com.google.protobuf.FieldMask;
 import io.spine.client.OrderBy;
+import io.spine.client.ResponseFormat;
 import io.spine.server.entity.EntityRecord;
 import io.spine.server.entity.storage.CompositeQueryParameter;
 import io.spine.server.entity.storage.QueryParameters;
@@ -47,9 +48,8 @@
 /**
  * An {@code Entity} lookup using {@linkplain QueryParameters Spine query parameters}.
  *
- * @implNote A single {@linkplain #find(QueryParameters, FieldMask) find()} call may turn
- *         into several
- *         Datastore reads. See {@link DsFilters} for details.
+ * @implNote A single {@linkplain #find(QueryParameters, ResponseFormat) find()} call may
+ *         turn into several Datastore reads. See {@link DsFilters} for details.
  */
 final class DsLookupByQueries {
 
@@ -68,30 +68,34 @@ final class DsLookupByQueries {
 
     /**
      * Finds a collection of entities matching provided {@link QueryParameters} in Datastore and
-     * returns them one-by-one applying a {@code fieldMask}.
+     * returns them according to the specified {@code format}.
      *
      * @param params
-     *         parameters specifying the filters, order and limit for records to conform to
-     * @param fieldMask
-     *         a mask to apply to each returned {@linkplain EntityRecord#getState() records state}
+     *         parameters specifying the filters for records to conform to
+     * @param format
+     *         format of the search specifying limits, order and field mask to apply to the results
      * @return an iterator over the entity records from the Datastore
      */
-    Iterator<EntityRecord> find(QueryParameters params, FieldMask fieldMask) {
-        List<StructuredQuery<Entity>> queries = splitToMultipleDsQueries(params);
+    Iterator<EntityRecord> find(QueryParameters params, ResponseFormat format) {
+        List<StructuredQuery<Entity>> queries = splitToMultipleDsQueries(params, format);
+        FieldMask mask = format.getFieldMask();
         if (queries.size() == 1) {
-            Iterator<EntityRecord> results = find(queries.get(0), fieldMask);
+            Iterator<EntityRecord> results = find(queries.get(0), mask);
             return results;
         }
-        Iterator<EntityRecord> results = find(queries, params.orderBy(), params.limit(), fieldMask);
+
+        Iterator<EntityRecord> results =
+                find(queries, format.getOrderBy(), format.getLimit(), mask);
         return results;
     }
 
-    private List<StructuredQuery<Entity>> splitToMultipleDsQueries(QueryParameters params) {
+    private List<StructuredQuery<Entity>>
+    splitToMultipleDsQueries(QueryParameters params, ResponseFormat format) {
         checkNotNull(params);
 
         List<StructuredQuery<Entity>> queries = buildDsFilters(params.iterator())
                 .stream()
-                .map(new QueryWithFilter(params, Kind.of(typeUrl)))
+                .map(new QueryWithFilter(format, Kind.of(typeUrl)))
                 .collect(toList());
         return queries;
     }

datastore/src/main/java/io/spine/server/storage/datastore/DsProjectionStorage.java

@@ -23,6 +23,7 @@
 import com.google.protobuf.FieldMask;
 import com.google.protobuf.Message;
 import com.google.protobuf.Timestamp;
+import io.spine.client.ResponseFormat;
 import io.spine.server.entity.EntityRecord;
 import io.spine.server.projection.Projection;
 import io.spine.server.projection.ProjectionStorage;
@@ -87,25 +88,15 @@ public boolean delete(I id) {
     }
 
     @Override
-    protected Iterator<EntityRecord> readMultipleRecords(Iterable<I> ids) {
-        return recordStorage().readMultiple(ids);
+    protected Iterator<EntityRecord> readAllRecords(ResponseFormat format) {
+        return recordStorage.readAllRecords(format);
     }
 
     @Override
     protected Iterator<EntityRecord> readMultipleRecords(Iterable<I> ids, FieldMask fieldMask) {
         return recordStorage().readMultiple(ids, fieldMask);
     }
 
-    @Override
-    protected Iterator<EntityRecord> readAllRecords() {
-        return recordStorage().readAll();
-    }
-
-    @Override
-    protected Iterator<EntityRecord> readAllRecords(FieldMask fieldMask) {
-        return recordStorage().readAll(fieldMask);
-    }
-
     @Override
     public Iterator<I> index() {
         return recordStorage.index();

datastore/src/main/java/io/spine/server/storage/datastore/DsRecordStorage.java

@@ -28,6 +28,8 @@
 import com.google.protobuf.Any;
 import com.google.protobuf.FieldMask;
 import com.google.protobuf.Message;
+import io.spine.client.OrderBy;
+import io.spine.client.ResponseFormat;
 import io.spine.server.entity.EntityRecord;
 import io.spine.server.entity.storage.ColumnTypeRegistry;
 import io.spine.server.entity.storage.EntityQuery;
@@ -128,34 +130,23 @@ protected Optional<EntityRecord> readRecord(I id) {
     }
 
     @Override
-    protected Iterator<@Nullable EntityRecord> readMultipleRecords(Iterable<I> ids) {
-        return idLookup.findActive(ids);
-    }
-
-    @Override
-    protected Iterator<@Nullable EntityRecord> readMultipleRecords(Iterable<I> ids,
-                                                                   FieldMask fieldMask) {
-        return idLookup.findActive(ids, fieldMask);
-    }
-
-    @Override
-    protected Iterator<EntityRecord> readAllRecords() {
-        Iterator<EntityRecord> result = readAllRecords(FieldMask.getDefaultInstance());
+    protected Iterator<EntityRecord> readAllRecords(ResponseFormat format) {
+        Iterator<EntityRecord> result = queryLookup.find(activeEntityQueryParams(this), format);
         return result;
     }
 
     @Override
-    protected Iterator<EntityRecord> readAllRecords(FieldMask fieldMask) {
-        Iterator<EntityRecord> result = queryLookup.find(activeEntityQueryParams(this), fieldMask);
-        return result;
+    protected Iterator<EntityRecord> readAllRecords(EntityQuery<I> query, ResponseFormat format) {
+        if (isQueryForAll(query)) {
+            return readAll(format);
+        }
+        return queryBy(query, format);
     }
 
     @Override
-    protected Iterator<EntityRecord> readAllRecords(EntityQuery<I> query, FieldMask fieldMask) {
-        if (isQueryForAll(query)) {
-            return readAll(fieldMask);
-        }
-        return queryBy(query, fieldMask);
+    protected Iterator<@Nullable EntityRecord> readMultipleRecords(Iterable<I> ids,
+                                                                   FieldMask fieldMask) {
+        return idLookup.findActive(ids, fieldMask);
     }
 
     @SuppressWarnings("PMD.SimplifyBooleanReturns")
@@ -167,19 +158,7 @@ private static <I> boolean isQueryForAll(EntityQuery<I> query) {
         }
 
         QueryParameters params = query.getParameters();
-        if (notEmpty(params)) {
-            return false;
-        }
-
-        if (params.ordered()) {
-            return false;
-        }
-
-        if (params.limited()) {
-            return false;
-        }
-
-        return true;
+        return !notEmpty(params);
     }
 
     /**
@@ -190,17 +169,18 @@ private static <I> boolean isQueryForAll(EntityQuery<I> query) {
      *
      * @param entityQuery
      *         the {@link EntityQuery} to query the Datastore by
-     * @param fieldMask
-     *         the {@code FieldMask} to apply to all the retrieved entity states
+     * @param responseFormat
+     *         the {@code ResponseFormat} according to which the result is retrieved
      * @return an iterator over the resulting entity records
      */
-    private Iterator<EntityRecord> queryBy(EntityQuery<I> entityQuery, FieldMask fieldMask) {
+    private Iterator<EntityRecord> queryBy(EntityQuery<I> entityQuery,
+                                           ResponseFormat responseFormat) {
         EntityQuery<I> completeQuery = includeLifecycle(entityQuery);
         Collection<I> idFilter = completeQuery.getIds();
         QueryParameters params = completeQuery.getParameters();
         Iterator<EntityRecord> result = idFilter.isEmpty()
-                                        ? queryLookup.find(params, fieldMask)
-                                        : queryByIdsAndColumns(idFilter, params, fieldMask);
+                                        ? queryLookup.find(params, responseFormat)
+                                        : queryByIdsAndColumns(idFilter, params, responseFormat);
         return result;
     }
 
@@ -220,20 +200,23 @@ private EntityQuery<I> includeLifecycle(EntityQuery<I> entityQuery) {
      *         the IDs to search by
      * @param params
      *         the additional query parameters
-     * @param fieldMask
-     *         the {@code FieldMask} to apply to all the retrieved entity states
+     * @param format
+     *         the format of the response including a field mask to apply, response size limit
+     *         and ordering
      * @return an iterator over the resulting entity records
      */
     private Iterator<EntityRecord> queryByIdsAndColumns(Collection<I> acceptableIds,
                                                         QueryParameters params,
-                                                        FieldMask fieldMask) {
+                                                        ResponseFormat format) {
         Predicate<Entity> inMemPredicate = columnPredicate(params);
-        if (params.ordered()) {
-            if (params.limited()) {
-                return idLookup.find(acceptableIds, fieldMask, inMemPredicate,
-                                     params.orderBy(), params.limit());
+        FieldMask fieldMask = format.getFieldMask();
+        if (format.hasOrderBy()) {
+            OrderBy order = format.getOrderBy();
+            int limit = format.getLimit();
+            if (limit > 0) {
+                return idLookup.find(acceptableIds, fieldMask, inMemPredicate, order, limit);
             }
-            return idLookup.find(acceptableIds, fieldMask, inMemPredicate, params.orderBy());
+            return idLookup.find(acceptableIds, fieldMask, inMemPredicate, order);
         }
         return idLookup.find(acceptableIds, fieldMask, inMemPredicate);
     }

datastore/src/main/java/io/spine/server/storage/datastore/QueryWithFilter.java

@@ -24,7 +24,7 @@
 import com.google.cloud.datastore.Query;
 import com.google.cloud.datastore.StructuredQuery;
 import io.spine.client.OrderBy;
-import io.spine.server.entity.storage.QueryParameters;
+import io.spine.client.ResponseFormat;
 
 import java.util.function.Function;
 
@@ -41,17 +41,18 @@ final class QueryWithFilter implements Function<StructuredQuery.Filter, Structur
 
     private final StructuredQuery.Builder<Entity> builder;
 
-    QueryWithFilter(QueryParameters params, Kind kind) {
-        checkNotNull(params);
+    QueryWithFilter(ResponseFormat format, Kind kind) {
+        checkNotNull(format);
         checkNotNull(kind);
 
         this.builder = Query.newEntityQueryBuilder()
                             .setKind(kind.getValue());
-        if (params.ordered()) {
-            this.builder.setOrderBy(translateOrderBy(params.orderBy()));
+        if (format.hasOrderBy()) {
+            this.builder.setOrderBy(translateOrderBy(format.getOrderBy()));
         }
-        if (params.limited()) {
-            this.builder.setLimit(params.limit());
+        int limit = format.getLimit();
+        if (limit > 0) {
+            this.builder.setLimit(limit);
         }
     }
 

datastore/src/test/java/io/spine/server/storage/datastore/BigDataTester.java

@@ -21,6 +21,7 @@
 package io.spine.server.storage.datastore;
 
 import com.google.common.base.Throwables;
+import io.spine.client.ResponseFormat;
 import io.spine.logging.Logging;
 import io.spine.server.entity.EntityRecord;
 import io.spine.server.entity.storage.EntityRecordWithColumns;
@@ -120,7 +121,7 @@ public void testBigDataOperations(RecordStorage<I> storage) {
         long readStart = System.currentTimeMillis();
 
         // Do not test data equality here, only the sizes and time
-        Iterator<EntityRecord> readRecords = storage.readAll();
+        Iterator<EntityRecord> readRecords = storage.readAll(ResponseFormat.getDefaultInstance());
 
         long readEnd = System.currentTimeMillis();
         long readTime = readEnd - readStart;