Add caching signal enrichment provider to configs

Signed-off-by: David Schwilk <david.schwilk@bosch.io>
eclipse-ditto · Oct 25, 2021 · 486d663 · 486d663
1 parent 3772fef
commit 486d663
Show file tree

Hide file tree

Showing 6 changed files with 50 additions and 20 deletions.
diff --git a/connectivity/service/src/main/resources/connectivity.conf b/connectivity/service/src/main/resources/connectivity.conf
@@ -549,6 +549,8 @@ ditto {
           expire-after-create = ${?CONNECTIVITY_SIGNAL_ENRICHMENT_CACHE_EXPIRE_AFTER_CREATE}
         }
       }
+      # Which caching signal enrichment facade is used by the signal-enrichment.provider
+      caching-signal-enrichment-facade.provider = ${?CONNECTIVITY_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
     }
 
     persistence-ping {

diff --git a/gateway/service/src/main/resources/gateway.conf b/gateway/service/src/main/resources/gateway.conf
@@ -21,7 +21,7 @@ ditto {
       actor-props-factory = "org.eclipse.ditto.gateway.service.endpoints.actors.DefaultHttpRequestActorPropsFactory"
 
       # headers to read the proxy-protocol from for HTTPS enforcement
-      protocol-headers = [ "X-Forwarded-Proto", "x_forwarded_proto" ]
+      protocol-headers = ["X-Forwarded-Proto", "x_forwarded_proto"]
 
       forcehttps = false
       forcehttps = ${?FORCE_HTTPS}
@@ -81,6 +81,8 @@ ditto {
           expire-after-create = 2m
           expire-after-create = ${?GATEWAY_SIGNAL_ENRICHMENT_CACHE_EXPIRE_AFTER_CREATE}
         }
+        # The caching signal enrichment provider to be used
+        caching-signal-enrichment-facade.provider =  ${?GATEWAY_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
       }
 
       acknowledgement {

diff --git a/...rg/eclipse/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacade.java b/...rg/eclipse/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacade.java
@@ -52,29 +52,47 @@
  * Instantiated once per cluster node so that it builds up a cache across all signal enrichments on a local cluster
  * node.
  */
-public final class DittoCachingSignalEnrichmentFacade implements CachingSignalEnrichmentFacade{
+public final class DittoCachingSignalEnrichmentFacade implements CachingSignalEnrichmentFacade {
 
     private static final ThreadSafeDittoLogger LOGGER = DittoLoggerFactory
             .getThreadSafeLogger(DittoCachingSignalEnrichmentFacade.class);
     private static final String CACHE_NAME_SUFFIX = "_signal_enrichment_cache";
 
     private final Cache<CacheKey, JsonObject> extraFieldsCache;
 
-    DittoCachingSignalEnrichmentFacade(
-            final SignalEnrichmentFacade cacheLoaderFacade,
+    private DittoCachingSignalEnrichmentFacade(final SignalEnrichmentFacade cacheLoaderFacade,
             final CacheConfig cacheConfig,
             final Executor cacheLoaderExecutor,
             final String cacheNamePrefix) {
 
-        final var cacheLoader =
-                SignalEnrichmentCacheLoader.of(checkNotNull(cacheLoaderFacade, "cacheLoaderFacade"));
-        final var cacheName = checkNotNull(cacheNamePrefix, "cacheNamePrefix") + CACHE_NAME_SUFFIX;
+        final var cacheLoader = SignalEnrichmentCacheLoader.of(cacheLoaderFacade);
+        final var cacheName = cacheNamePrefix + CACHE_NAME_SUFFIX;
 
         extraFieldsCache = CacheFactory.createCache(
                 cacheLoader,
-                checkNotNull(cacheConfig, "cacheConfig"),
+                cacheConfig,
                 cacheName,
-                checkNotNull(cacheLoaderExecutor, "cacheLoaderExecutor"));
+                cacheLoaderExecutor);
+    }
+
+    /**
+     * Returns a new {@code DittoCachingSignalEnrichmentFacade} instance.
+     *
+     * @param cacheLoaderFacade the facade whose argument-result-pairs we are caching.
+     * @param cacheConfig the cache configuration to use for the cache.
+     * @param cacheLoaderExecutor the executor to use in order to asynchronously load cache entries.
+     * @param cacheNamePrefix the prefix to use as cacheName of the cache.
+     * @throws NullPointerException if any argument is null.
+     */
+    public static DittoCachingSignalEnrichmentFacade newInstance(final SignalEnrichmentFacade cacheLoaderFacade,
+            final CacheConfig cacheConfig,
+            final Executor cacheLoaderExecutor,
+            final String cacheNamePrefix) {
+
+        return new DittoCachingSignalEnrichmentFacade(checkNotNull(cacheLoaderFacade, "cacheLoaderFacade"),
+                checkNotNull(cacheConfig, "cacheConfig"),
+                checkNotNull(cacheLoaderExecutor, "cacheLoaderExecutor"),
+                checkNotNull(cacheNamePrefix, "cacheNamePrefix"));
     }
 
     @Override

diff --git a/...se/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacadeProvider.java b/...se/ditto/internal/models/signalenrichment/DittoCachingSignalEnrichmentFacadeProvider.java
@@ -36,7 +36,7 @@ public CachingSignalEnrichmentFacade getSignalEnrichmentFacade(final SignalEnric
             final Executor cacheLoaderExecutor,
             final String cacheNamePrefix) {
 
-        return new DittoCachingSignalEnrichmentFacade(cacheLoaderFacade, cacheConfig, cacheLoaderExecutor,
+        return DittoCachingSignalEnrichmentFacade.newInstance(cacheLoaderFacade, cacheConfig, cacheLoaderExecutor,
                 cacheNamePrefix);
     }
 

diff --git a/thingsearch/service/src/main/resources/things-search.conf b/thingsearch/service/src/main/resources/things-search.conf
@@ -55,6 +55,10 @@ ditto {
       enabled = ${?INDEX_INITIALIZATION_ENABLED}
     }
 
+    signal-enrichment {
+      caching-signal-enrichment-facade.provider =  ${?THINGS_SEARCH_CACHING_SIGNAL_ENRICHMENT_PROVIDER}
+    }
+
     updater {
       max-idle-time = 15m
       max-idle-time = ${?ACTIVITY_CHECK_INTERVAL}

diff --git a/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/EnforcementFlowTest.java b/...rg/eclipse/ditto/thingsearch/service/persistence/write/streaming/EnforcementFlowTest.java
@@ -45,7 +45,9 @@
 import org.eclipse.ditto.thingsearch.service.persistence.write.model.ThingWriteModel;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
+import org.junit.FixMethodOrder;
 import org.junit.Test;
+import org.junit.runners.MethodSorters;
 
 import com.typesafe.config.ConfigFactory;
 
@@ -62,11 +64,13 @@
 import akka.stream.testkit.javadsl.TestSource;
 import akka.testkit.TestProbe;
 import akka.testkit.javadsl.TestKit;
+import scala.concurrent.duration.Duration;
 import scala.concurrent.duration.FiniteDuration;
 
 /**
- * Unit tests for {@link EnforcementFlow}.
+ * Unit tests for {@link EnforcementFlow}. Contains fix method order to allow for longer setup during first test.
  */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
 public final class EnforcementFlowTest {
 
     private static ActorSystem system;
@@ -104,7 +108,7 @@ public void updateThingAndPolicyRevisions() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -155,7 +159,7 @@ public void ignoreCacheWhenRequestedToUpdate() {
 
         final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
         final EnforcementFlow underTest =
-                EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                         system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
         materializeTestProbes(underTest.create(false, 1));
@@ -231,7 +235,7 @@ public void computeThingCacheValueFromThingEvents() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -244,7 +248,7 @@ public void computeThingCacheValueFromThingEvents() {
             sourceProbe.sendComplete();
 
             // WHEN: policy is retrieved with up-to-date revisions
-            policiesProbe.expectMsgClass(SudoRetrievePolicy.class);
+            policiesProbe.expectMsgClass(Duration.apply(30, TimeUnit.SECONDS), SudoRetrievePolicy.class);
             final var policy = Policy.newBuilder(policyId).setRevision(1).build();
             policiesProbe.reply(SudoRetrievePolicyResponse.of(policyId, policy, DittoHeaders.empty()));
 
@@ -292,7 +296,7 @@ public void forceRetrieveThing() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -337,7 +341,7 @@ public void eventSequenceNumberTooLow() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -381,7 +385,7 @@ public void eventMissed() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -424,7 +428,7 @@ public void noInitialCreatedOrDeletedEvent() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));
@@ -491,7 +495,7 @@ public void onlyApplyRelevantEvents() {
 
             final StreamConfig streamConfig = DefaultStreamConfig.of(ConfigFactory.empty());
             final EnforcementFlow underTest =
-                    EnforcementFlow.of(streamConfig, thingsProbe.ref(), policiesProbe.ref(),
+                    EnforcementFlow.of(system, streamConfig, thingsProbe.ref(), policiesProbe.ref(),
                             system.dispatchers().defaultGlobalDispatcher(), system.getScheduler());
 
             materializeTestProbes(underTest.create(false, 1));