Improve Connection API

CHANGELOG.md

@@ -106,7 +106,7 @@
 ### Features
 
 * **api:** Add new configuration provider which enables default values ([a973cd1](https://github.com/wuespace/telestion-core/commit/a973cd1f0d30513bcfcae655db156138f74b145a))
-* **connection:** Add configuration of baud rate to `SerialConn` verticle. Usages of the `de.wuespace.telestion.services.connection.rework.serial.SerialConn` now requires an additional parameter: `int baudRate`. To migrate, add this parameter to any application configuration using this verticle. The value used before is `9600`. Therefore, you can use `"baudRate": 9600` to match the old default configuration. ([ae4dad2](https://github.com/wuespace/telestion-core/commit/ae4dad2c9732047551ea74cca5b35b45bfd47f83))
+* **connection:** Add configuration of baud rate to `SerialConn` verticle. Usages of the `de.wuespace.telestion.services.connection.serial.SerialConn` now requires an additional parameter: `int baudRate`. To migrate, add this parameter to any application configuration using this verticle. The value used before is `9600`. Therefore, you can use `"baudRate": 9600` to match the old default configuration. ([ae4dad2](https://github.com/wuespace/telestion-core/commit/ae4dad2c9732047551ea74cca5b35b45bfd47f83))
 
 ## [0.3.0](https://github.com/wuespace/telestion-core/compare/v0.2.1...v0.3.0) (2021-05-08)
 

modules/telestion-api/build.gradle

@@ -1,6 +1,6 @@
 application {
     // Define the main class for the application.
-    mainClass.set('de.wuespace.telestion.application.Application')
+    mainClass.set('de.wuespace.telestion.application.Telestion')
 }
 
 java {

modules/telestion-application/build.gradle

@@ -1,6 +1,6 @@
 application {
     // Define the main class for the application.
-    mainClass.set('de.wuespace.telestion.application.Application')
+    mainClass.set('de.wuespace.telestion.application.Telestion')
 }
 
 java {

modules/telestion-application/src/main/java/de/wuespace/telestion/application/Telestion.java

@@ -5,7 +5,9 @@
 import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
+
 import java.util.Collections;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import de.wuespace.telestion.services.config.Configuration;
@@ -16,9 +18,6 @@
  * @author Jan von Pichowski
  */
 public final class Telestion extends AbstractVerticle {
-
-	private static final Logger logger = LoggerFactory.getLogger(Telestion.class);
-
 	/**
 	 * Deploys this Telestion verticle.
 	 *
@@ -38,18 +37,30 @@ public void start(Promise<Void> startPromise) {
 				startPromise.fail(configRes.cause());
 				return;
 			}
-			var conf = configRes.result().getJsonObject("org.telestion.configuration").mapTo(Configuration.class);
-			conf.verticles().stream().flatMap(c -> Collections.nCopies(c.magnitude(), c).stream()).forEach(v -> {
-				logger.info("Deploying {}", v.name());
-				var future = vertx.deployVerticle(v.verticle(), new DeploymentOptions().setConfig(v.jsonConfig()));
-				future.onFailure(Throwable::printStackTrace);
-			});
+
+			var conf = configRes
+					.result()
+					.getJsonObject("de.wuespace.telestion.configuration")
+					.mapTo(Configuration.class);
+
+			conf.verticles().stream()
+					.flatMap(c -> Collections.nCopies(c.magnitude(), c).stream()).forEach(v -> {
+						logger.info("Deploying {}", v.name());
+						var future = vertx.deployVerticle(
+								v.verticle(),
+								new DeploymentOptions().setConfig(v.jsonConfig())
+						);
+						future.onFailure(Throwable::printStackTrace);
+					});
+
 			startPromise.complete();
 		});
 	}
 
 	@Override
-	public void stop(Promise<Void> stopPromise) throws Exception {
-
+	public void stop(Promise<Void> stopPromise) {
+		stopPromise.complete();
 	}
+
+	private static final Logger logger = LoggerFactory.getLogger(Telestion.class);
 }

modules/telestion-application/src/main/java/de/wuespace/telestion/launcher/Launcher.java

@@ -1,9 +1,13 @@
 package de.wuespace.telestion.launcher;
 
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Handler;
 import io.vertx.core.Verticle;
 import io.vertx.core.Vertx;
+
 import java.time.Duration;
 import java.util.Arrays;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,19 +40,13 @@ public static void main(String[] args) {
 	public static void start(String... verticleNames) {
 		logger.info("Deploying {} verticles", verticleNames.length);
 		var vertx = Vertx.vertx();
-		// vertx.eventBus().registerDefaultCodec(Position.class, JsonMessageCodec.Instance(Position.class));
-		Arrays.stream(verticleNames).forEach(verticleName -> {
-			logger.info("Deploying verticle {}", verticleName);
-			vertx.setPeriodic(Duration.ofSeconds(5).toMillis(), timerId -> {
-				vertx.deployVerticle(verticleName, res -> {
-					if (res.failed()) {
-						logger.error("Failed to deploy verticle {} retrying in 5s", verticleName, res.cause());
-						return;
-					}
-					logger.info("Deployed verticle {} with id {}", verticleName, res.result());
-					vertx.cancelTimer(timerId);
-				});
-			});
+
+		Arrays.stream(verticleNames).forEach(name -> {
+			logger.info("Deploying verticle {}", name);
+			vertx.setPeriodic(
+					Duration.ofSeconds(5).toMillis(),
+					timerId -> vertx.deployVerticle(name, deploymentHandler(vertx, name, timerId))
+			);
 		});
 	}
 
@@ -61,19 +59,27 @@ public static void start(String... verticleNames) {
 	public static void start(Verticle... verticles) {
 		logger.info("Deploying {} verticles", verticles.length);
 		var vertx = Vertx.vertx();
-		// vertx.eventBus().registerDefaultCodec(Position.class, JsonMessageCodec.Instance(Position.class));
-		Arrays.stream(verticles).forEach(verticleName -> {
-			logger.info("Deploying verticle {}", verticleName);
-			vertx.setPeriodic(Duration.ofSeconds(5).toMillis(), timerId -> {
-				vertx.deployVerticle(verticleName, res -> {
-					if (res.failed()) {
-						logger.error("Failed to deploy verticle {} retrying in 5s", verticleName, res.cause());
-						return;
-					}
-					logger.info("Deployed verticle {} with id {}", verticleName, res.result());
-					vertx.cancelTimer(timerId);
-				});
-			});
+
+		Arrays.stream(verticles).forEach(instance -> {
+			logger.info("Deploying verticle {}", instance);
+			vertx.setPeriodic(
+					Duration.ofSeconds(5).toMillis(),
+					timerId -> vertx.deployVerticle(
+							instance,
+							deploymentHandler(vertx, instance.getClass().getName(), timerId)
+					)
+			);
 		});
 	}
+
+	private static Handler<AsyncResult<String>> deploymentHandler(Vertx vertx, String verticle, Long timerId) {
+		return res -> {
+			if (res.failed()) {
+				logger.error("Failed to deploy verticle {} retrying in 5s", verticle, res.cause());
+				return;
+			}
+			logger.info("Deployed verticle {} with id {}", verticle, res.result());
+			vertx.cancelTimer(timerId);
+		};
+	}
 }

modules/telestion-services/build.gradle

@@ -1,6 +1,6 @@
 application {
     // Define the main class for the application.
-    mainClass.set('de.wuespace.telestion.application.Application')
+    mainClass.set('de.wuespace.telestion.application.Telestion')
 }
 
 java {
@@ -19,8 +19,6 @@ ext {
 description = 'Services and service components that are re-usable in any Telestion Project Application'
 
 dependencies {
-    // implementation name: 'engine'
-
     implementation project(':modules:telestion-api')
 
     implementation group: 'com.fazecast', name: 'jSerialComm', version: '2.9.1'

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/Broadcaster.java

@@ -0,0 +1,103 @@
+package de.wuespace.telestion.services.connection;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import de.wuespace.telestion.api.config.Config;
+import de.wuespace.telestion.api.message.JsonMessage;
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Promise;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+public final class Broadcaster extends AbstractVerticle {
+
+	public final static int NO_BROADCASTING = -1;
+	public final static int DEFAULT_ID = 0;
+
+	public final record Configuration(@JsonProperty
+									  String inAddress,
+									  @JsonProperty
+									  int id) implements JsonMessage {
+		public Configuration() {
+			this(null, DEFAULT_ID);
+		}
+	}
+
+	@Override
+	public void start(Promise<Void> startPromise) {
+		this.config = Config.get(this.config, new Configuration(), this.config(), Configuration.class);
+
+		if (broadcasterMap.containsKey(this.config.id())) {
+			startPromise.fail("The broadcasters id #%d is already taken!".formatted(this.config.id()));
+			return;
+		}
+
+		broadcasterMap.put(this.config.id(), this);
+
+		startPromise.complete();
+	}
+
+	@Override
+	public void stop(Promise<Void> stopPromise) {
+		stopPromise.complete();
+	}
+
+	/**
+	 * Registers a new address for the broadcaster with the given id if it was previously specified in the config.
+	 *
+	 * @param broadcasterId		0 = default broadcaster, must be >= 0
+	 * @param address			Address on VertX bus it must be sent to
+	 * @return		if registration process was successful
+	 */
+	public static boolean register(int broadcasterId, String address) {
+		if (broadcasterId == NO_BROADCASTING) {
+			return true;
+		}
+
+		if (!broadcasterMap.containsKey(broadcasterId) || broadcasterId < 0) {
+			logger.error("Setup invalid!");
+			return false;
+		}
+
+		var broadcaster = broadcasterMap.get(broadcasterId);
+		broadcaster.addressList.add(address);
+
+		broadcaster.getVertx().eventBus()
+				.consumer(broadcaster.config.inAddress(), raw -> {
+					if (!JsonMessage.on(RawMessage.class, raw, broadcaster::send)) {
+						if (!JsonMessage.on(SenderData.class,
+								raw,
+								msg -> broadcaster.send(new RawMessage(msg.rawData())))) {
+							JsonMessage.on(ConnectionData.class,
+									raw,
+									msg -> broadcaster.send(new RawMessage(msg.rawData())));
+						}
+					}
+				});
+		return true;
+	}
+
+	public String[] getAddresses() {
+		return addressList.toArray(String[]::new);
+	}
+
+	public Broadcaster() {
+		this(null);
+	}
+
+	public Broadcaster(Configuration config) {
+		this.config = config;
+		this.addressList = new HashSet<>();
+	}
+
+	private void send(RawMessage msg) {
+		addressList.forEach(addr -> vertx.eventBus().publish(addr, msg.json()));
+	}
+
+	private Configuration config;
+	private final Set<String> addressList;
+
+	private static final Map<Integer, Broadcaster> broadcasterMap = new HashMap<>();
+	private static Logger logger = LoggerFactory.getLogger(Broadcaster.class);
+}

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/ConnectionData.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/ConnectionData.java

@@ -1,4 +1,4 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import de.wuespace.telestion.api.message.JsonMessage;

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/ConnectionDetails.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/ConnectionDetails.java

@@ -1,4 +1,4 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import de.wuespace.telestion.api.message.JsonMessage;

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/IpDetails.java

@@ -0,0 +1,13 @@
+package de.wuespace.telestion.services.connection;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import de.wuespace.telestion.api.message.JsonMessage;
+
+public record IpDetails(@JsonProperty
+						String ip,
+						@JsonProperty
+						int port) implements JsonMessage {
+	public IpDetails() {
+		this("0.0.0.0", 0);
+	}
+}

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/RawMessage.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/RawMessage.java

@@ -1,10 +1,9 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import de.wuespace.telestion.api.message.JsonMessage;
 
 public record RawMessage(@JsonProperty byte[] data) implements JsonMessage {
-	@SuppressWarnings("unused")
 	private RawMessage() {
 		this(null);
 	}

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/Receiver.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/Receiver.java

@@ -1,4 +1,4 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.vertx.core.AbstractVerticle;

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/Sender.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/Sender.java

@@ -1,4 +1,4 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import de.wuespace.telestion.api.config.Config;

modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/rework/SenderData.java → modules/telestion-services/src/main/java/de/wuespace/telestion/services/connection/SenderData.java

@@ -1,4 +1,4 @@
-package de.wuespace.telestion.services.connection.rework;
+package de.wuespace.telestion.services.connection;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import de.wuespace.telestion.api.message.JsonMessage;