Add extension to allow adding custom command forwarding

e.g. for commands issued by a route which is provided by a CustomApiRoutesProvider

Signed-off-by: Yannic Klem <Yannic.Klem@bosch.io>

connectivity/service/src/main/resources/connectivity.conf

@@ -963,6 +963,7 @@ akka {
 }
 
 include "ditto-protocol-subscriber.conf"
+include "ditto-edge-api.conf"
 
 akka-contrib-mongodb-persistence-connection-journal {
   class = "akka.contrib.persistence.mongodb.MongoJournal"

edge/api/pom.xml

@@ -60,7 +60,6 @@
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-thingsearch-model</artifactId>
         </dependency>
-
         <dependency>
             <groupId>org.eclipse.ditto</groupId>
             <artifactId>ditto-thingsearch-api</artifactId>

edge/api/src/main/java/org/eclipse/ditto/edge/api/dispatching/EdgeCommandForwarderActor.java

@@ -76,7 +76,10 @@ public static Props props(final ActorRef pubSubMediator, final ShardRegions shar
 
     @Override
     public Receive createReceive() {
-        return ReceiveBuilder.create()
+        final Receive receiveExtension = EdgeCommandForwarderExtension.get(context().system())
+                .getReceiveExtension(getContext());
+
+        final Receive forwardingReceive = ReceiveBuilder.create()
                 .match(MessageCommand.class, this::forwardToThings)
                 .match(ThingCommand.class, this::forwardToThings)
                 .match(RetrieveThings.class, this::forwardToThingsAggregator)
@@ -92,6 +95,8 @@ public Receive createReceive() {
                 )
                 .matchAny(m -> log.warning("Got unknown message: {}", m))
                 .build();
+
+        return receiveExtension.orElse(forwardingReceive);
     }
 
     private void forwardToThings(final MessageCommand<?, ?> messageCommand) {

edge/api/src/main/java/org/eclipse/ditto/edge/api/dispatching/EdgeCommandForwarderExtension.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.edge.api.dispatching;
+
+import static org.eclipse.ditto.base.model.common.ConditionChecker.checkNotNull;
+
+import org.eclipse.ditto.base.service.DittoExtensionPoint;
+
+import akka.actor.AbstractActor;
+import akka.actor.ActorSystem;
+
+/**
+ * This extension allows to extend commands that are forwarded from the edges of ditto to other microservices by
+ * handling them in the {@link AbstractActor.Receive} provided by {@link #getReceiveExtension(akka.actor.AbstractActor.ActorContext)}.
+ */
+public interface EdgeCommandForwarderExtension extends DittoExtensionPoint {
+
+    static EdgeCommandForwarderExtension get(final ActorSystem actorSystem) {
+        checkNotNull(actorSystem, "actorSystem");
+        return ExtensionId.INSTANCE.get(actorSystem);
+    }
+
+    /**
+     * Builds the receive extension to allow custom handling of messages.
+     * This receive will be applied BEFORE the default receives of {@link EdgeCommandForwarderActor}, so it's possible
+     * to overwrite the default handling.
+     *
+     * @param actorContext can be used for example to determine the original sender of a message.
+     * @return The desired receive extension.
+     */
+    AbstractActor.Receive getReceiveExtension(AbstractActor.ActorContext actorContext);
+
+    final class ExtensionId extends DittoExtensionPoint.ExtensionId<EdgeCommandForwarderExtension> {
+
+        private static final String CONFIG_PATH = "ditto.edge-command-forwarder-extension";
+        private static final ExtensionId INSTANCE = new ExtensionId(EdgeCommandForwarderExtension.class);
+
+        private ExtensionId(final Class<EdgeCommandForwarderExtension> parentClass) {
+            super(parentClass);
+        }
+
+        @Override
+        protected String getConfigPath() {
+            return CONFIG_PATH;
+        }
+
+    }
+
+}

edge/api/src/main/java/org/eclipse/ditto/edge/api/dispatching/NoOpEdgeCommandForwarderExtension.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.edge.api.dispatching;
+
+import akka.actor.AbstractActor;
+import akka.japi.pf.ReceiveBuilder;
+
+final class NoOpEdgeCommandForwarderExtension implements EdgeCommandForwarderExtension{
+
+    @Override
+    public AbstractActor.Receive getReceiveExtension(final AbstractActor.ActorContext actorContext) {
+        return ReceiveBuilder.create().build();
+    }
+
+}

edge/api/src/main/resources/ditto-edge-api.conf

@@ -0,0 +1,2 @@
+ditto.edge-command-forwarder-extension = org.eclipse.ditto.edge.api.dispatching.NoOpEdgeCommandForwarderExtension
+ditto.edge-command-forwarder-extension = ${?DITTO_EDGE_COMMAND_FORWARDER_EXTENSION}

gateway/service/src/main/resources/gateway.conf

@@ -539,6 +539,7 @@ akka {
 }
 
 include "ditto-protocol-subscriber.conf"
+include "ditto-edge-api.conf"
 
 authentication-dispatcher {
   type = Dispatcher