Allow custom configuration for the SseConnectionSupervisor extension

Signed-off-by: Yannic Klem <yannic.klem@bosch.io>
eclipse-ditto · Jul 12, 2022 · 3894dc6 · 3894dc6
1 parent 8945fc7
commit 3894dc6
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 12 deletions.
diff --git a/.../java/org/eclipse/ditto/gateway/service/endpoints/routes/sse/SseConnectionSupervisor.java b/.../java/org/eclipse/ditto/gateway/service/endpoints/routes/sse/SseConnectionSupervisor.java
@@ -14,9 +14,13 @@
 
 import static org.eclipse.ditto.base.model.common.ConditionChecker.checkNotNull;
 
+import org.eclipse.ditto.base.service.DittoExtensionIds;
 import org.eclipse.ditto.base.service.DittoExtensionPoint;
+import org.eclipse.ditto.base.service.RootChildActorStarter;
 import org.eclipse.ditto.gateway.service.streaming.actors.StreamSupervisor;
 
+import com.typesafe.config.Config;
+
 import akka.actor.ActorSystem;
 
 /**
@@ -33,18 +37,26 @@ public interface SseConnectionSupervisor extends DittoExtensionPoint, StreamSupe
      * @throws NullPointerException if {@code actorSystem} is {@code null}.
      * @since 3.0.0
      */
-    static SseConnectionSupervisor get(final ActorSystem actorSystem) {
+    static SseConnectionSupervisor get(final ActorSystem actorSystem, final Config config) {
         checkNotNull(actorSystem, "actorSystem");
-        return ExtensionId.INSTANCE.get(actorSystem);
+        checkNotNull(config, "config");
+        final var extensionIdConfig = ExtensionId.computeConfig(config);
+        return DittoExtensionIds.get(actorSystem)
+                .computeIfAbsent(extensionIdConfig, ExtensionId::new)
+                .get(actorSystem);
     }
 
     final class ExtensionId extends DittoExtensionPoint.ExtensionId<SseConnectionSupervisor> {
 
-        private static final String CONFIG_PATH = "ditto.gateway.streaming.sse.connection-supervisor";
-        private static final ExtensionId INSTANCE = new ExtensionId(SseConnectionSupervisor.class);
+        private static final String CONFIG_KEY = "sse-connection-supervisor";
+        private static final String CONFIG_PATH = "ditto.extensions." + CONFIG_KEY;
+
+        private ExtensionId(final ExtensionIdConfig<SseConnectionSupervisor> extensionIdConfig) {
+            super(extensionIdConfig);
+        }
 
-        private ExtensionId(final Class<SseConnectionSupervisor> parentClass) {
-            super(parentClass);
+        static ExtensionIdConfig<SseConnectionSupervisor> computeConfig(final Config config) {
+            return ExtensionIdConfig.of(SseConnectionSupervisor.class, config, CONFIG_KEY);
         }
 
         @Override

diff --git a/...in/java/org/eclipse/ditto/gateway/service/endpoints/routes/sse/ThingsSseRouteBuilder.java b/...in/java/org/eclipse/ditto/gateway/service/endpoints/routes/sse/ThingsSseRouteBuilder.java
@@ -153,12 +153,14 @@ private ThingsSseRouteBuilder(final ActorSystem actorSystem,
         this.streamingConfig = streamingConfig;
         this.queryFilterCriteriaFactory = queryFilterCriteriaFactory;
         this.pubSubMediator = pubSubMediator;
+
         final Config config = actorSystem.settings().config();
         final var dittoExtensionsConfig = ScopedConfig.dittoExtension(config);
         eventSniffer = SseEventSniffer.get(actorSystem, dittoExtensionsConfig);
+        sseConnectionSupervisor = SseConnectionSupervisor.get(actorSystem, dittoExtensionsConfig);
+
         final var sseConfig = ScopedConfig.getOrEmpty(config, "ditto.gateway.streaming.sse");
         sseAuthorizationEnforcer = StreamingAuthorizationEnforcer.get(actorSystem, sseConfig);
-        sseConnectionSupervisor = SseConnectionSupervisor.get(actorSystem);
     }
 
     /**

diff --git a/gateway/service/src/main/resources/gateway.conf b/gateway/service/src/main/resources/gateway.conf
@@ -55,6 +55,8 @@ ditto {
     outgoing-websocket-event-sniffer = org.eclipse.ditto.gateway.service.endpoints.routes.websocket.NoOpOutgoingWebSocketEventSniffer
     # The provider for custom API routes.
     custom-api-routes-provider = org.eclipse.ditto.gateway.service.endpoints.routes.NoopCustomApiRoutesProvider
+    # The supervisor for SSE connections
+    sse-connection-supervisor = org.eclipse.ditto.gateway.service.endpoints.routes.sse.NoOpSseConnectionSupervisor
   }
 
   service-name = "gateway"
@@ -185,10 +187,6 @@ ditto {
 
         # The provider enforcer for SSE connections
         streaming-authorization-enforcer = "org.eclipse.ditto.gateway.service.streaming.NoOpAuthorizationEnforcer"
-
-        # The supervisor for SSE connections
-        connection-supervisor = "org.eclipse.ditto.gateway.service.endpoints.routes.sse.NoOpSseConnectionSupervisor"
-        connection-supervisor = ${?GATEWAY_SSE_CONNECTION_SUPERVISOR}
       }
     }
 

diff --git a/gateway/service/src/test/resources/test.conf b/gateway/service/src/test/resources/test.conf
@@ -146,6 +146,7 @@ ditto {
     incoming-websocket-event-sniffer = org.eclipse.ditto.gateway.service.endpoints.routes.websocket.NoOpIncomingWebSocketEventSniffer
     outgoing-websocket-event-sniffer = org.eclipse.ditto.gateway.service.endpoints.routes.websocket.NoOpOutgoingWebSocketEventSniffer
     custom-api-routes-provider = org.eclipse.ditto.gateway.service.endpoints.routes.NoopCustomApiRoutesProvider
+    sse-connection-supervisor = org.eclipse.ditto.gateway.service.endpoints.routes.sse.NoOpSseConnectionSupervisor
   }
 
   mapping-strategy.implementation = "org.eclipse.ditto.gateway.service.util.GatewayMappingStrategies"
@@ -162,7 +163,6 @@ ditto {
     streaming {
       sse {
         streaming-authorization-enforcer = org.eclipse.ditto.gateway.service.streaming.NoOpAuthorizationEnforcer
-        connection-supervisor = "org.eclipse.ditto.gateway.service.endpoints.routes.sse.NoOpSseConnectionSupervisor"
       }
       websocket {
         # the max queue size of how many inflight Commands a single Websocket client can have