feat: Allow config file to be provided at runtime

* feat: Allow config file to be provided at runtime

* feat: Handle promise completion in start methods

* feat: Update README for new properties file

* fix: Don't use future for synchronous code

Signed-off-by: Katherine Stanley <katheris@uk.ibm.com>

Author: katheris

README.md

@@ -35,5 +35,7 @@ Test out the app by connecting to the websocket endpoints (UI will be added soon
  - Disconnect from the consume websocket
 
 ## Configuration
-To configure the application to connect to your Kafka edit the properties file in `src/main/resources`.
-If your Kafka is secured you will need to enable the security configuration options and add your certificate to `src/main/resources`.
+To configure the application to connect to your Kafka edit the properties file called `kafka.properties`.
+Alternatively you can provide a custom path to the properties file using `-Dproperties_path=<path>` when starting the application.
+
+If your Kafka is secured you will need to enable the security configuration options in your properties file

src/main/resources/kafka.properties kafka.propertiessrc/main/resources/kafka.properties renamed to kafka.properties

@@ -9,16 +9,17 @@
 import io.vertx.config.ConfigRetrieverOptions;
 import io.vertx.config.ConfigStoreOptions;
 import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.TimeoutStream;
 import io.vertx.core.eventbus.Message;
+import io.vertx.core.file.FileSystem;
 import io.vertx.core.json.JsonObject;
 import io.vertx.kafka.client.producer.KafkaProducer;
 import io.vertx.kafka.client.producer.KafkaProducerRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.time.Instant;
 import java.util.HashMap;
 import java.util.Optional;
 
@@ -29,18 +30,34 @@ public class PeriodicProducer extends AbstractVerticle {
 
   @Override
   public void start(Promise<Void> startPromise) {
-    ConfigRetriever.create(vertx,
+    loadKafkaConfig()
+      .onSuccess(config -> {
+        setup(config);
+        startPromise.complete();
+      })
+      .onFailure(startPromise::fail);
+  }
+
+  private Future<JsonObject> loadKafkaConfig() {
+    String path = Optional.ofNullable(System.getProperty("properties_path")).orElse("kafka.properties");
+    ConfigRetriever configRetriever =  ConfigRetriever.create(vertx,
       new ConfigRetrieverOptions().addStore(
         new ConfigStoreOptions()
           .setType("file")
           .setFormat("properties")
-          .setConfig(new JsonObject().put("path", "kafka.properties").put("raw-data", true))))
-      .getConfig()
-      .onSuccess(config -> setup(config, startPromise))
-      .onFailure(startPromise::fail);
+          .setConfig(new JsonObject().put("path", path).put("raw-data", true))));
+    FileSystem fileSystem = vertx.fileSystem();
+    return fileSystem.exists(path)
+      .compose(exists -> {
+        if (exists) {
+          return configRetriever.getConfig();
+        } else {
+          return Future.failedFuture("Kafka properties file is missing. Either specify using -Dproperties_path=<path> or use the default path of kafka.properties.");
+        }
+      });
   }
 
-  private void setup(JsonObject config, Promise<Void> startPromise) {
+  private void setup(JsonObject config) {
     HashMap<String, String> props = new HashMap<>();
     config.forEach(entry -> props.put(entry.getKey(), entry.getValue().toString()));
     KafkaProducer<String, String> kafkaProducer = KafkaProducer.create(vertx, props);
@@ -55,7 +72,6 @@ private void setup(JsonObject config, Promise<Void> startPromise) {
 
     vertx.eventBus().<JsonObject>consumer(Main.PERIODIC_PRODUCER_ADDRESS, message -> handleCommand(timerStream, message));
     logger.info("🚀 PeriodicProducer started");
-    startPromise.complete();
   }
 
   private void handleCommand(TimeoutStream timerStream, Message<JsonObject> message) {

src/main/java/kafka/vertx/demo/PeriodicProducer.java

@@ -9,9 +9,12 @@
 import io.vertx.config.ConfigRetrieverOptions;
 import io.vertx.config.ConfigStoreOptions;
 import io.vertx.core.AbstractVerticle;
+import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.eventbus.EventBus;
 import io.vertx.core.eventbus.MessageConsumer;
+import io.vertx.core.file.FileSystem;
+import io.vertx.core.http.HttpServer;
 import io.vertx.core.http.ServerWebSocket;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.Router;
@@ -35,38 +38,43 @@ public class WebSocketServer extends AbstractVerticle {
 
   @Override
   public void start(Promise<Void> startPromise) {
-    loadKafkaConfig(startPromise);
-
     Router router = Router.router(vertx);
     router.get().handler(StaticHandler.create());
 
-    vertx.createHttpServer()
-      .requestHandler(router)
-      .webSocketHandler(this::handleWebSocket)
-      .listen(8080, "localhost")
-      .onSuccess(ok -> {
-        logger.info("🚀 WebSocketServer started");
-        startPromise.complete();
-      })
-      .onFailure(err -> {
-        logger.error("❌ WebSocketServer failed to listen", err);
-        startPromise.fail(err);
-      });
+    loadKafkaConfig()
+      .compose(config -> startWebSocket(router, config))
+      .onSuccess(ok -> startPromise.complete())
+      .onFailure(startPromise::fail);
   }
 
-  private void loadKafkaConfig(Promise<Void> startPromise) {
-    ConfigRetriever.create(vertx,
+  private Future<JsonObject> loadKafkaConfig() {
+    String path = Optional.ofNullable(System.getProperty("properties_path")).orElse("kafka.properties");
+    ConfigRetriever configRetriever =  ConfigRetriever.create(vertx,
       new ConfigRetrieverOptions().addStore(
         new ConfigStoreOptions()
           .setType("file")
           .setFormat("properties")
-          .setConfig(new JsonObject().put("path", "kafka.properties").put("raw-data", true))))
-      .getConfig()
-      .onSuccess(config -> {
-        kafkaConfig = new HashMap<>();
-        config.forEach(entry -> kafkaConfig.put(entry.getKey(), entry.getValue().toString()));
-      })
-      .onFailure(startPromise::fail);
+          .setConfig(new JsonObject().put("path", path).put("raw-data", true))));
+    FileSystem fileSystem = vertx.fileSystem();
+    return fileSystem.exists(path)
+      .compose(exists -> {
+        if (exists) {
+          return configRetriever.getConfig();
+        } else {
+          return Future.failedFuture("Kafka properties file is missing. Either specify using -Dproperties_path=<path> or use the default path of kafka.properties.");
+        }
+      });
+  }
+
+  private Future<HttpServer> startWebSocket(Router router, JsonObject config) {
+    kafkaConfig = new HashMap<>();
+    config.forEach(entry -> kafkaConfig.put(entry.getKey(), entry.getValue().toString()));
+    return vertx.createHttpServer()
+      .requestHandler(router)
+      .webSocketHandler(this::handleWebSocket)
+      .listen(8080, "localhost")
+      .onSuccess(ok -> logger.info("🚀 WebSocketServer started"))
+      .onFailure(err -> logger.error("❌ WebSocketServer failed to listen", err));
   }
 
   private void handleWebSocket(ServerWebSocket webSocket) {