Fix caching issue in composite-spaces

Changes in the base space were not immediately reflected in the composite-space.
Also fix test class UpdateSpaceWithExtensionApiIT, so that it always deletes the composite-space before deleting the base-space.

Signed-off-by: Benjamin Rögner <benjamin.roegner@here.com>

Author: roegi

xyz-hub-service/src/main/java/com/here/xyz/hub/connectors/models/Space.java

@@ -83,6 +83,10 @@ public class Space extends com.here.xyz.models.hub.Space implements Cloneable {
   @JsonIgnore
   private Map<ConnectorType, Map<String, List<ResolvableListenerConnectorRef>>> resolvedConnectorRefs;
 
+  public static Future<Space> resolveSpace(Marker marker, String spaceId) {
+    return Service.spaceConfigClient.get(marker, spaceId);
+  }
+
   public static Future<Connector> resolveConnector(Marker marker, String connectorId) {
     Promise<Connector> p = Promise.promise();
     resolveConnector(marker, connectorId, p);
@@ -290,4 +294,13 @@ public Map<String,Object> asMap() {
       return Collections.emptyMap();
     }
   }
+
+  /**
+   * Used for logging purposes.
+   * @return
+   */
+  @Override
+  public String toString() {
+    return Json.encode(this);
+  }
 }

xyz-hub-service/src/main/java/com/here/xyz/hub/task/FeatureTask.java

@@ -25,6 +25,7 @@
 import static io.netty.handler.codec.http.HttpResponseStatus.NOT_FOUND;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.hash.Hasher;
 import com.google.common.hash.Hashing;
 import com.here.xyz.events.CountFeaturesEvent;
 import com.here.xyz.events.DeleteFeaturesByTagEvent;
@@ -81,6 +82,11 @@ public abstract class FeatureTask<T extends Event<?>, X extends FeatureTask<T, ?
    */
   public Space space;
 
+  /**
+   * The spaces being extended by {@link #space} (if existing).
+   */
+  public List<Space> extendedSpaces;
+
   /**
    * The storage connector to be used for this operation.
    */
@@ -144,13 +150,15 @@ public String getCacheKey() {
     }
     try {
       //noinspection UnstableApiUsage
-      cacheKey = Hashing.murmur3_128().newHasher()
+      Hasher hasher = Hashing.murmur3_128().newHasher()
           .putString(getEvent().getCacheString(), Charset.defaultCharset())
           .putString(responseType.toString(), Charset.defaultCharset())
-          .putLong(space.contentUpdatedAt)
-          .hash()
-          .toString();
-      return cacheKey;
+          .putLong(space.contentUpdatedAt);
+
+      if (space.getExtension() != null)
+        extendedSpaces.forEach(extendedSpace -> hasher.putLong(extendedSpace.getContentUpdatedAt()));
+
+      return cacheKey = hasher.hash().toString();
     } catch (JsonProcessingException e) {
       return null;
     }

xyz-hub-service/src/main/java/com/here/xyz/hub/task/FeatureTaskHandler.java

@@ -86,6 +86,7 @@
 import com.here.xyz.models.geojson.implementation.FeatureCollection;
 import com.here.xyz.models.geojson.implementation.FeatureCollection.ModificationFailure;
 import com.here.xyz.models.geojson.implementation.XyzNamespace;
+import com.here.xyz.models.hub.Space.Extension;
 import com.here.xyz.responses.BinaryResponse;
 import com.here.xyz.responses.CountResponse;
 import com.here.xyz.responses.ErrorResponse;
@@ -98,9 +99,11 @@
 import com.here.xyz.responses.SuccessResponse;
 import com.here.xyz.responses.XyzResponse;
 import io.vertx.core.AsyncResult;
+import io.vertx.core.CompositeFuture;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.MultiMap;
+import io.vertx.core.Promise;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.Cookie;
 import io.vertx.core.json.DecodeException;
@@ -746,65 +749,103 @@ private static EventNotification createNotification(NotificationContext nc, Payl
    * Resolves the space, its storage and its listeners.
    */
   static <X extends FeatureTask> void resolveSpace(final X task, final Callback<X> callback) {
+    try {
+      resolveSpace(task)
+          .compose(space -> CompositeFuture.all(
+              resolveStorageConnector(task),
+              resolveListenersAndProcessors(task),
+              resolveExtendedSpaces(task, space)
+          ))
+          .onFailure(t -> callback.exception(t))
+          .onSuccess(connector -> callback.call(task));
+    }
+    catch (Exception e) {
+      callback.exception(new HttpException(INTERNAL_SERVER_ERROR, "Unable to load the resource definition.", e));
+    }
+  }
+
+  private static <X extends FeatureTask> Future<Space> resolveSpace(final X task) {
     try {
       //FIXME: Can be removed once the Space events are handled by the SpaceTaskHandler (refactoring pending ...)
-      if (task.space != null) { //If the space is already given we don't need to retrieve it
-        onSpaceResolved(task, callback);
-        return;
-      }
+      if (task.space != null) //If the space is already given we don't need to retrieve it
+        return Future.succeededFuture(task.space);
 
       //Load the space definition.
-      Service.spaceConfigClient.get(task.getMarker(), task.getEvent().getSpace())
-          .onFailure(t -> {
-            logger.warn(task.getMarker(), "Unable to load the space definition for space '{}' {}", task.getEvent().getSpace(), t);
-            callback.exception(new HttpException(INTERNAL_SERVER_ERROR, "Unable to load the resource definition", t));
-          })
-          .onSuccess(space -> {
-            task.space = space;
-            if (task.space != null)
-              task.getEvent().setParams(task.space.getStorage().getParams());
-            onSpaceResolved(task, callback);
-          });
+      return Space.resolveSpace(task.getMarker(), task.getEvent().getSpace())
+          .compose(
+              space -> {
+                task.space = space;
+                if (space != null)
+                  task.getEvent().setParams(space.getStorage().getParams());
+                return Future.succeededFuture(space);
+              },
+              t -> {
+                logger.warn(task.getMarker(), "Unable to load the space definition for space '{}' {}", task.getEvent().getSpace(), t);
+                return Future.failedFuture(new HttpException(INTERNAL_SERVER_ERROR, "Unable to load the resource definition", t));
+              }
+          );
     }
     catch (Exception e) {
-      callback.exception(new HttpException(INTERNAL_SERVER_ERROR, "Unable to load the resource definition.", e));
+      return Future.failedFuture(new HttpException(INTERNAL_SERVER_ERROR, "Unable to load the resource definition.", e));
     }
   }
 
-  private static <X extends FeatureTask> void onSpaceResolved(final X task, final Callback<X> callback) {
-    if (task.space == null) {
-      callback.exception(new HttpException(NOT_FOUND, "The resource with this ID does not exist."));
-      return;
-    }
-    logger.debug(task.getMarker(), "Given space configuration is: {}", Json.encode(task.space));
+  private static <X extends FeatureTask> Future<Space> resolveExtendedSpaces(X task, Space extendingSpace) {
+    if (extendingSpace == null)
+      return Future.succeededFuture();
+    return resolveExtendedSpace(task, extendingSpace.getExtension());
+  }
 
-    final String storageId = task.space.getStorage().getId();
-    AbstractHttpServerVerticle.addStreamInfo(task.context, "SID", storageId);
-    Space.resolveConnector(task.getMarker(), storageId, (arStorage) -> {
-      if (arStorage.failed()) {
-        callback.exception(new InvalidStorageException("Unable to load the definition for this storage."));
-        return;
-      }
-      task.storage = arStorage.result();
-      onStorageResolved(task, callback);
-    });
+  private static <X extends FeatureTask> Future<Space> resolveExtendedSpace(X task, Extension spaceExtension) {
+    if (spaceExtension == null)
+      return Future.succeededFuture();
+    return Space.resolveSpace(task.getMarker(), spaceExtension.getSpaceId())
+        .compose(
+            extendedSpace -> {
+              if (task.extendedSpaces == null)
+                task.extendedSpaces = new ArrayList();
+              task.extendedSpaces.add(extendedSpace);
+              return resolveExtendedSpace(task, extendedSpace.getExtension()); //Go to next extension level
+            },
+            t -> Future.failedFuture(t)
+        );
   }
 
-  private static <X extends FeatureTask> void onStorageResolved(final X task, final Callback<X> callback) {
+  private static <X extends FeatureTask> Future<Connector> resolveStorageConnector(final X task) {
+    if (task.space == null)
+      return Future.failedFuture(new HttpException(NOT_FOUND, "The resource with this ID does not exist."));
+
+    logger.debug(task.getMarker(), "Given space configuration is: {}", task.space);
+
+    final String storageId = task.space.getStorage().getId();
+    AbstractHttpServerVerticle.addStreamInfo(task.context, "SID", storageId);
+    return Space.resolveConnector(task.getMarker(), storageId)
+        .compose(
+            connector -> {
+              task.storage = connector;
+              return Future.succeededFuture(connector);
+            },
+            t -> Future.failedFuture(new InvalidStorageException("Unable to load the definition for this storage."))
+        );
+  }
+
+  private static <X extends FeatureTask> Future<Void> resolveListenersAndProcessors(final X task) {
+    Promise<Void> p = Promise.promise();
     try {
       //Also resolve all listeners & processors
       CompletableFuture.allOf(
           resolveConnectors(task.getMarker(), task.space, ConnectorType.LISTENER),
           resolveConnectors(task.getMarker(), task.space, ConnectorType.PROCESSOR)
       ).thenRun(() -> {
         //All listener & processor refs have been resolved now
-        callback.call(task);
+        p.complete();
       });
     }
     catch (Exception e) {
       logger.error(task.getMarker(), "The listeners for this space cannot be initialized", e);
-      callback.exception(new HttpException(INTERNAL_SERVER_ERROR, "The listeners for this space cannot be initialized"));
+      p.fail(new HttpException(INTERNAL_SERVER_ERROR, "The listeners for this space cannot be initialized"));
     }
+    return p.future();
   }
 
   private static CompletableFuture<Void> resolveConnectors(Marker marker, final Space space, final ConnectorType connectorType) {

xyz-hub-test/src/test/java/com/here/xyz/hub/rest/TTLTestsIT.java

@@ -79,6 +79,12 @@ public void testCacheWithGeoJson() {
         .then()
         .extract().body().path("features[0].id");
 
+    try {
+      //Wait 1s to give the asynchronous cache-write some time.
+      Thread.sleep(1_000);
+    }
+    catch (InterruptedException e) {}
+
     String id_cached = given()
         .headers(getAuthHeaders(AuthProfile.ACCESS_ALL))
         .contentType(APPLICATION_JSON)

xyz-hub-test/src/test/java/com/here/xyz/hub/rest/UpdateSpaceWithExtensionApiIT.java

@@ -33,16 +33,16 @@ public class UpdateSpaceWithExtensionApiIT extends TestSpaceWithFeature {
 
   @BeforeClass
   public static void setupClass() {
-    removeSpace("x-psql-test");
     removeSpace("x-psql-test-extension");
+    removeSpace("x-psql-test");
     createSpace();
     createSpaceWithExtension();
   }
 
   @AfterClass
   public static void tearDownClass() {
-    removeSpace("x-psql-test");
     removeSpace("x-psql-test-extension");
+    removeSpace("x-psql-test");
   }
 
   private static void createSpaceWithExtension() {