Feature/multiple datasinks

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/loader/FileConfigurationLoader.java

@@ -95,6 +95,7 @@ public Object loadConfiguration() {
 	 * file path and the resource loader
 	 * @return
 	 */
+	@SuppressWarnings("resource")
 	private InputStreamReader getJsonReader() {
 
 		InputStream stream = null;

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/core/AbstractRouteCreator.java

@@ -23,9 +23,10 @@
  * SPDX-License-Identifier: MIT
  ******************************************************************************/
 
-
 package org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core;
 
+import java.util.Map;
+
 import org.apache.camel.builder.RouteBuilder;
 
 public abstract class AbstractRouteCreator implements IRouteCreator {
@@ -54,10 +55,17 @@ public void addRouteToRouteBuilder(RouteConfiguration routeConfig) {
 		String dataSourceEndpoint = RouteCreatorHelper.getDataSourceEndpoint(routesConfiguration, routeConfig.getDatasource());
 		String[] dataSinkEndpoints = RouteCreatorHelper.getDataSinkEndpoints(routesConfiguration, routeConfig.getDatasinks());
 		String[] dataTransformerEndpoints = RouteCreatorHelper.getDataTransformerEndpoints(routesConfiguration, routeConfig.getTransformers());
+		Map<String, String[]> datasinkMapping = RouteCreatorHelper.getDataSinkMapping(routesConfiguration, routeConfig.getDatasinkMappingConfiguration());
 		String routeId = routeConfig.getRouteId();
 
-		configureRoute(routeConfig, dataSourceEndpoint, dataSinkEndpoints, dataTransformerEndpoints, routeId);
+		if (datasinkMapping == null || datasinkMapping.isEmpty()) {
+			configureRoute(routeConfig, dataSourceEndpoint, dataSinkEndpoints, dataTransformerEndpoints, routeId);
+		} else {
+			configureRoute(routeConfig, dataSourceEndpoint, dataSinkEndpoints, dataTransformerEndpoints, datasinkMapping, routeId);
+		}
 	}
 
 	protected abstract void configureRoute(RouteConfiguration routeConfig, String dataSourceEndpoint, String[] dataSinkEndpoints, String[] dataTransformerEndpoints, String routeId);
+
+	protected abstract void configureRoute(RouteConfiguration routeConfig, String dataSourceEndpoint, String[] dataSinkEndpoints, String[] dataTransformerEndpoints, Map<String, String[]> DataSinkMapping, String routeId);
 }

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/core/RouteConfiguration.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (C) 2021 the Eclipse BaSyx Authors
+ * Copyright (C) 2024 the Eclipse BaSyx Authors
  * 
  * Permission is hereby granted, free of charge, to any person obtaining
  * a copy of this software and associated documentation files (the
@@ -29,24 +29,63 @@
 import java.util.List;
 import java.util.Map;
 
+/**
+ * @author DataBridge authors, jungjan
+ */
 public class RouteConfiguration {
 	private String trigger;
 	private String routeId;
 	private String datasource;
 	private List<String> transformers = new ArrayList<>();
 	private List<String> datasinks = new ArrayList<>();
+	private Map<String, String[]> datasinkMappingConfiguration;
 
 	private Map<String, Object> triggerData = new HashMap<>();
 
 	public RouteConfiguration() {
 	}
 
 	/**
+	 * Constructs a new RouteConfiguration object with a mapping configuration to
+	 * map distinct transformators to multiple datasinks.
+	 *
 	 * @param trigger
+	 *            the trigger for the route configuration
 	 * @param routeId
+	 *            the ID of the route
 	 * @param datasource
+	 *            the datasource associated with the route
 	 * @param transformers
+	 *            the list of transformers to be applied in the route
 	 * @param datasinks
+	 *            the list of datasinks to which data should be routed
+	 * @param datasinkMappingConfiguration
+	 *            the mapping configuration for datasinks, mapping each datasink to
+	 *            its corresponding configuration
+	 */
+	public RouteConfiguration(String trigger, String datasource, List<String> transformers, List<String> datasinks, Map<String, String[]> datasinkMappingConfiguration) {
+		this.trigger = trigger;
+		this.datasource = datasource;
+		this.transformers = transformers;
+		this.datasinks = datasinks;
+		this.datasinkMappingConfiguration = datasinkMappingConfiguration;
+	}
+
+	/**
+	 * Constructs a new RouteConfiguration object without a mapping configuration to
+	 * map distinct transformators to multiple datasinks. I.e., all transformators
+	 * would be equally applied to all data sinks.
+	 *
+	 * @param trigger
+	 *            the trigger for the route configuration
+	 * @param routeId
+	 *            the ID of the route (optional, can be null)
+	 * @param datasource
+	 *            the datasource associated with the route
+	 * @param transformers
+	 *            the list of transformers to be applied in the route
+	 * @param datasinks
+	 *            the list of datasinks to which data should be routed
 	 */
 	public RouteConfiguration(String trigger, String datasource, List<String> transformers, List<String> datasinks) {
 		this.trigger = trigger;
@@ -56,7 +95,7 @@ public RouteConfiguration(String trigger, String datasource, List<String> transf
 	}
 
 	public RouteConfiguration(RouteConfiguration configuration) {
-		this(configuration.getRouteTrigger(), configuration.getDatasource(), configuration.getTransformers(), configuration.getDatasinks());
+		this(configuration.getRouteTrigger(), configuration.getDatasource(), configuration.getTransformers(), configuration.getDatasinks(), configuration.getDatasinkMappingConfiguration());
 		setRouteId(configuration.getRouteId());
 		this.triggerData = configuration.triggerData;
 	}
@@ -97,4 +136,12 @@ public void setDatasource(String datasource) {
 		this.datasource = datasource;
 	}
 
+	public Map<String, String[]> getDatasinkMappingConfiguration() {
+		return datasinkMappingConfiguration;
+	}
+
+	public void setDatasinkMappingConfiguration(Map<String, String[]> datasinkMappingConfiguration) {
+		this.datasinkMappingConfiguration = datasinkMappingConfiguration;
+	}
+
 }

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/core/RouteCreatorHelper.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (C) 2021 the Eclipse BaSyx Authors
+ * Copyright (C) 2024 the Eclipse BaSyx Authors
  * 
  * Permission is hereby granted, free of charge, to any person obtaining
  * a copy of this software and associated documentation files (the
@@ -24,35 +24,57 @@
  ******************************************************************************/
 package org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core;
 
-import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+/**
+ * @author DataBridge authors, jungjan
+ */
 public class RouteCreatorHelper {
 	private RouteCreatorHelper() {
 	}
 
 	public static String getDataSourceEndpoint(RoutesConfiguration routesConfiguration, String dataSourceId) {
-		return routesConfiguration.getDatasources().get(dataSourceId).getConnectionURI();
+		return routesConfiguration.getDatasources()
+				.get(dataSourceId)
+				.getConnectionURI();
 	}
 
 	public static String getDataSinkEndpoint(RoutesConfiguration routesConfiguration, String dataSinkId) {
-		return routesConfiguration.getDatasinks().get(dataSinkId).getConnectionURI();
+		return routesConfiguration.getDatasinks()
+				.get(dataSinkId)
+				.getConnectionURI();
 	}
 
 	public static String[] getDataSinkEndpoints(RoutesConfiguration routesConfiguration, List<String> dataSinkIdList) {
-		List<String> endpoints = new ArrayList<>();
-		for (String dataSinkId : dataSinkIdList) {
-			endpoints.add(routesConfiguration.getDatasinks().get(dataSinkId).getConnectionURI());
-		}
+		return dataSinkIdList.stream()
+				.map(routesConfiguration.getDatasinks()::get)
+				.map(dataSinkConfiguration -> dataSinkConfiguration.getConnectionURI())
+				.toArray(String[]::new);
+
+	}
 
-		return endpoints.toArray(new String[0]);
+	public static String[] getDataTransformerEndpoints(RoutesConfiguration routesConfiguration, List<String> transformerIdLists) {
+		return transformerIdLists.stream()
+				.map(routesConfiguration.getTransformers()::get)
+				.map(dataTransformerConfiguration -> dataTransformerConfiguration.getConnectionURI())
+				.toArray(String[]::new);
 	}
 
-	public static String[] getDataTransformerEndpoints(RoutesConfiguration routesConfiguration, List<String> transformerIdList) {
-		List<String> endpoints = new ArrayList<>();
-		for (String transformerId : transformerIdList) {
-			endpoints.add(routesConfiguration.getTransformers().get(transformerId).getConnectionURI());
+	public static Map<String, String[]> getDataSinkMapping(RoutesConfiguration routesConfiguration, Map<String, String[]> datasinkMappingConfiguration) {
+		if (datasinkMappingConfiguration == null || datasinkMappingConfiguration.isEmpty()) {
+			return null;
 		}
-		return endpoints.toArray(new String[0]);
+		Set<String> dataSinkIds = datasinkMappingConfiguration.keySet();
+		Map<String, String> resolvedDataSinkEndpoints = dataSinkIds.stream()
+				.collect(Collectors.toMap(dataSinkId -> dataSinkId, dataSinkId -> routesConfiguration.getDatasinks()
+						.get(dataSinkId)
+						.getConnectionURI()));
+
+		return dataSinkIds.stream()
+				.collect(Collectors.toMap(resolvedDataSinkEndpoints::get, dataSinkId -> getDataTransformerEndpoints(routesConfiguration, Arrays.asList(datasinkMappingConfiguration.get(dataSinkId)))));
 	}
 }

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/event/EventRouteConfiguration.java

@@ -25,11 +25,12 @@
 package org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.event;
 
 import java.util.List;
+import java.util.Map;
 
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core.RouteConfiguration;
 
 /**
- * A connection of a single route (source, transformer(s), sink(s))
+ * A connection of a single route (source, transformer(s), sink(s), sinkmapper(s)
  *
  * @author haque, fischer
  *
@@ -40,6 +41,10 @@ public class EventRouteConfiguration extends RouteConfiguration {
 	public EventRouteConfiguration(String datasource, List<String> transformers, List<String> datasinks) {
 		super(ROUTE_TRIGGER, datasource, transformers, datasinks);
 	}
+
+	public EventRouteConfiguration(String datasource, List<String> transformers, List<String> datasinks, Map<String, String[]> datasinkMapping) {
+		super(ROUTE_TRIGGER, datasource, transformers, datasinks, datasinkMapping);
+	}
 
 	public EventRouteConfiguration(RouteConfiguration configuration) {
 		super(configuration);

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/event/EventRouteCreator.java

@@ -24,27 +24,60 @@
  ******************************************************************************/
 package org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.event;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
 import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.MulticastDefinition;
 import org.apache.camel.model.RouteDefinition;
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core.AbstractRouteCreator;
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core.RouteConfiguration;
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core.RoutesConfiguration;
 
 public class EventRouteCreator extends AbstractRouteCreator {
-
 	public EventRouteCreator(RouteBuilder routeBuilder, RoutesConfiguration routesConfiguration) {
 		super(routeBuilder, routesConfiguration);
 	}
 
 	@Override
 	protected void configureRoute(RouteConfiguration routeConfiguration, String dataSourceEndpoint, String[] dataSinkEndpoints, String[] dataTransformerEndpoints, String routeId) {
-		RouteDefinition routeDefinition = getRouteBuilder().from(dataSourceEndpoint).routeId(routeId).to("log:" + routeId);
+		RouteDefinition routeDefinition = startRouteDefinition(dataSourceEndpoint, routeId);
 
 		if (!(dataTransformerEndpoints == null || dataTransformerEndpoints.length == 0)) {
-			routeDefinition.to(dataTransformerEndpoints).to("log:" + routeId);
+			routeDefinition.to(dataTransformerEndpoints)
+					.to("log:" + routeId);
 		}
 
-		routeDefinition.to(dataSinkEndpoints[0]).to("log:" + routeId);
+		routeDefinition.to(dataSinkEndpoints)
+				.to("log:" + routeId);
+	}
+
+	@Override
+	protected void configureRoute(RouteConfiguration routeConfig, String dataSourceEndpoint, String[] dataSinkEndpoints, String[] dataTransformerEndpoints, Map<String, String[]> dataSinkMapping, String routeId) {
+		MulticastDefinition routeDefinition = startRouteDefinition(dataSourceEndpoint, routeId).multicast();
+		dataSinkMapping.forEach((dataSink, dataTransformers) -> routeDefinition.pipeline()
+				.to(dataTransformers)
+				.to(dataSink)
+				.to("log:" + routeId));
+
+		getUnmappedEndpoints(dataSinkEndpoints, dataSinkMapping).forEach(dataSink -> routeDefinition.to(dataSink)
+				.to("log: " + routeId));
+
+		routeDefinition.end();
 	}
 
-}
+	private List<String> getUnmappedEndpoints(String[] dataSinkEndpoints, Map<String, String[]> dataSinkMapping) {
+		return Arrays.stream(dataSinkEndpoints)
+				.filter(Predicate.not(dataSinkMapping::containsKey))
+				.collect(Collectors.toList());
+	}
+
+	private RouteDefinition startRouteDefinition(String dataSourceEndpoint, String routeId) {
+		return getRouteBuilder().from(dataSourceEndpoint)
+				.routeId(routeId)
+				.to("log:" + routeId);
+	}
+}

databridge.core/src/main/java/org/eclipse/digitaltwin/basyx/databridge/core/configuration/route/request/RequestRouteConfiguration.java

@@ -25,6 +25,7 @@
 package org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.request;
 
 import java.util.List;
+import java.util.Map;
 
 import org.eclipse.digitaltwin.basyx.databridge.core.configuration.route.core.RouteConfiguration;
 import org.springframework.http.HttpMethod;
@@ -53,6 +54,10 @@ public RequestRouteConfiguration(String datasource, List<String> transformers, L
 		super(ROUTE_TRIGGER, datasource, transformers, datasinks);
 	}
 
+	public RequestRouteConfiguration(String datasource, List<String> transformers, List<String> datasinks, Map<String, String[]> datasinkMapping) {
+		super(ROUTE_TRIGGER, datasource, transformers, datasinks, datasinkMapping);
+	}
+
 	public RequestRouteConfiguration(RouteConfiguration configuration) {
 		super(configuration);
 		host = (String) getTriggerData().get(HOST);
@@ -85,7 +90,6 @@ public void setPort(String port) {
 	}
 
 	public String getRequestEndpointURI() {
-		return REQUEST_COMPONENT + ":" + REQUEST_PROTOCOL + "://" + getHost() + ":" + getPort() + getPath() + "?"
-				+ HTTP_METHOD_RESTRICT_PARAMETER;
+		return REQUEST_COMPONENT + ":" + REQUEST_PROTOCOL + "://" + getHost() + ":" + getPort() + getPath() + "?" + HTTP_METHOD_RESTRICT_PARAMETER;
 	}
 }