messagechannel.send() implementation

controller/build.gradle

@@ -0,0 +1,11 @@
+plugins {
+ id 'com.flipkart.varadhi.java-library-conventions'
+}
+
+dependencies {
+ api(project(":entities"))
+ api(project(":spi"))
+ api(project(":core"))
+
+ implementation("io.vertx:vertx-micrometer-metrics")
+}

controller/src/main/java/com/flipkart/varadhi/controller/ControllerMgr.java

@@ -0,0 +1,25 @@
+package com.flipkart.varadhi.controller;
+
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionOperation;
+import com.flipkart.varadhi.core.ophandlers.ControllerOpHandler;
+import com.flipkart.varadhi.core.ophandlers.ServerOpHandler;
+import com.flipkart.varadhi.core.proxies.ServerOpMgrProxy;
+import com.flipkart.varadhi.entities.VaradhiSubscription;
+import com.flipkart.varadhi.spi.db.MetaStore;
+import io.micrometer.core.instrument.MeterRegistry;
+
+import java.util.concurrent.CompletableFuture;
+
+public class ControllerMgr implements ControllerOpHandler {
+ private ServerOpHandler serverOpHandler;
+
+ public ControllerMgr(ServerOpHandler serverOpHandler) {
+ this.serverOpHandler = serverOpHandler;
+ }
+ @Override
+ public CompletableFuture<Void> StartSubscription(SubscriptionOperation operation) {
+ operation.markInProgress();
+ serverOpHandler.update(operation);
+ return CompletableFuture.completedFuture(null);
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/MessageChannel.java

@@ -0,0 +1,50 @@
+package com.flipkart.varadhi.core.cluster;
+
+
+import com.flipkart.varadhi.core.cluster.messages.*;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * This is subset of methods from vertx.EventBus which are node specific. Instead of using address concept, we will
+ * rely on path concept from http. address is not needed as this class is meant to represent a message channel to a specific
+ * node.
+ * publish methods are fire & forget.
+ * send methods don't have any response, but their delivery can be tracked using the future.
+ * request methods are for traditional request-response pattern.
+ *
+ * TODO: whether to use the ClusterMessage class. It is unlikely we will use reply-to-reply feature.
+ */
+public interface MessageChannel {
+
+ void publish(String path, ClusterMessage msg);
+
+ CompletableFuture<Void> send(String path, ClusterMessage msg);
+
+ CompletableFuture<ResponseMessage> request(String path, ClusterMessage msg);
+
+ // void addMessageHandler(String path, MessageHandler messageHandler);
+ <E extends ClusterMessage> void register(String address, Class<E> messageClazz, SendHandler<E> handler);
+
+ <E extends ClusterMessage> void register(String address, Class<E> messageClazz, RequestHandler<E> handler);
+
+ <E extends ClusterMessage> void register(String address, Class<E> messageClazz, PublishHandler<E> handler);
+
+ void removeMessageHandler(String path);
+
+ public static enum Method {
+ SEND("send"),
+ PUBLISH("publish"),
+ REQUEST("request");
+
+ private String name;
+
+ Method(String name) {
+ this.name = name;
+ }
+ @Override
+ public String toString() {
+ return name;
+ }
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/ClusterMessage.java

@@ -0,0 +1,18 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import lombok.Data;
+import lombok.Getter;
+
+@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "@messageType")
+@Getter
+public class ClusterMessage {
+ String id;
+ long timeStamp;
+
+ public ClusterMessage() {
+ this.id = java.util.UUID.randomUUID().toString();
+ this.timeStamp = System.currentTimeMillis();
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/PublishHandler.java

@@ -0,0 +1,5 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+public interface PublishHandler<E extends ClusterMessage> {
+ void handle(E message);
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/RequestHandler.java

@@ -0,0 +1,7 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface RequestHandler <E extends ClusterMessage> {
+ CompletableFuture<ResponseMessage> handle(E message);
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/ResponseMessage.java

@@ -0,0 +1,6 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+public class ResponseMessage extends ClusterMessage {
+ // indicates a response to a request message.
+ // should contain the reference to original request (e.g. request id).
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/SendHandler.java

@@ -0,0 +1,8 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+import java.util.concurrent.CompletableFuture;
+
+@FunctionalInterface
+public interface SendHandler<E extends ClusterMessage> {
+ CompletableFuture<Void> handle(E message);
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/SubscriptionMessage.java

@@ -0,0 +1,17 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+
+@Getter
+@EqualsAndHashCode(callSuper = true)
+public class SubscriptionMessage extends ClusterMessage {
+ private final String subscriptionId;
+ private final SubscriptionOperation operation;
+
+ public SubscriptionMessage(SubscriptionOperation operation) {
+ super();
+ this.subscriptionId = operation.getSubscriptionId();
+ this.operation = operation;
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/cluster/messages/SubscriptionOperation.java

@@ -0,0 +1,45 @@
+package com.flipkart.varadhi.core.cluster.messages;
+
+
+import com.flipkart.varadhi.entities.VaradhiSubscription;
+import lombok.Data;
+
+import java.util.UUID;
+
+@Data
+public class SubscriptionOperation {
+ public static enum Kind {
+ CREATE, START, STOP, DELETE, UPDATE
+ }
+ public static enum State {
+ SCHEDULED, ERRORED, COMPLETED, IN_PROGRESS
+ }
+
+ private String operationId;
+ private Kind kind;
+ private State state;
+ private String subscriptionId;
+ private String requestedBy;
+ private String errorMessage;
+ private long startTime;
+ private long endTime;
+
+ public void markInProgress() {
+ state = State.IN_PROGRESS;
+ }
+
+ public SubscriptionMessage toMessage() {
+ return new SubscriptionMessage(this);
+ }
+
+ public static SubscriptionOperation getSubscriptionOp(Kind opKind, String subscriptionId, String requestedBy) {
+ SubscriptionOperation op = new SubscriptionOperation();
+ op.setOperationId(UUID.randomUUID().toString());
+ op.setKind(opKind);
+ op.setState(State.SCHEDULED);
+ op.setSubscriptionId(subscriptionId);
+ op.setRequestedBy(requestedBy);
+ op.setStartTime(System.currentTimeMillis());
+ return op;
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/ophandlers/ControllerOpHandler.java

@@ -0,0 +1,9 @@
+package com.flipkart.varadhi.core.ophandlers;
+
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionOperation;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface ControllerOpHandler {
+ CompletableFuture<Void> StartSubscription(SubscriptionOperation operation);
+}

core/src/main/java/com/flipkart/varadhi/core/ophandlers/ServerOpHandler.java

@@ -0,0 +1,10 @@
+package com.flipkart.varadhi.core.ophandlers;
+
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionOperation;
+import com.flipkart.varadhi.entities.VaradhiSubscription;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface ServerOpHandler {
+ CompletableFuture<Void> update(SubscriptionOperation operation);
+}

core/src/main/java/com/flipkart/varadhi/core/proxies/ControllerMgrProxy.java

@@ -0,0 +1,27 @@
+package com.flipkart.varadhi.core.proxies;
+
+import com.flipkart.varadhi.core.cluster.MessageChannel;
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionMessage;
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionOperation;
+import com.flipkart.varadhi.core.ophandlers.ControllerOpHandler;
+
+import java.util.concurrent.CompletableFuture;
+
+public class ControllerMgrProxy implements ControllerOpHandler {
+ private final String CONTROLLER_ADDR = "Controller";
+ private final MessageChannel channel;
+
+ public ControllerMgrProxy(MessageChannel channel) {
+ this.channel = channel;
+ }
+
+ @Override
+ public CompletableFuture<Void> StartSubscription(SubscriptionOperation operation) {
+ SubscriptionMessage message = operation.toMessage();
+ return channel.send(getAddress(MessageChannel.Method.SEND, message), message);
+ }
+
+ private String getAddress(MessageChannel.Method method, SubscriptionMessage message) {
+ return CONTROLLER_ADDR + "." + message.getClass().getSimpleName() + "." + method;
+ }
+}

core/src/main/java/com/flipkart/varadhi/core/proxies/ServerOpMgrProxy.java

@@ -0,0 +1,27 @@
+package com.flipkart.varadhi.core.proxies;
+
+import com.flipkart.varadhi.core.cluster.MessageChannel;
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionMessage;
+import com.flipkart.varadhi.core.cluster.messages.SubscriptionOperation;
+import com.flipkart.varadhi.core.ophandlers.ServerOpHandler;
+import com.flipkart.varadhi.entities.VaradhiSubscription;
+
+import java.util.concurrent.CompletableFuture;
+
+public class ServerOpMgrProxy implements ServerOpHandler {
+ private final String SERVER_ADDR = "Server";
+ private final MessageChannel channel;
+
+ public ServerOpMgrProxy(MessageChannel channel) {
+ this.channel = channel;
+ }
+ @Override
+ public CompletableFuture<Void> update(SubscriptionOperation operation) {
+ SubscriptionMessage message = operation.toMessage();
+ return channel.send(getAddress(MessageChannel.Method.SEND, message), message);
+ }
+
+ private String getAddress(MessageChannel.Method method, SubscriptionMessage message) {
+ return SERVER_ADDR + "." + message.getClass().getSimpleName() + "." + method;
+ }
+}

server/build.gradle

@@ -14,6 +14,7 @@ dependencies {
  implementation(project(":messaging"))
  implementation(project(":entities"))
  implementation(project(":authz"))
+ implementation(project(":controller"))
  implementation(project(":consumer"))
 
  implementation("org.apache.logging.log4j:log4j-slf4j2-impl")
@@ -50,6 +51,8 @@ dependencies {
  implementation("org.apache.curator:curator-framework")
  runtimeOnly(project(":pulsar"))
 
+ testImplementation("io.vertx:vertx-zookeeper:4.5.1")
+
  testImplementation(project(":pulsar"))
  testImplementation(testFixtures(project(":spi")))
  testImplementation(testFixtures(project(":entities")))

server/src/main/java/com/flipkart/varadhi/CoreServices.java

@@ -1,7 +1,7 @@
 package com.flipkart.varadhi;
 
 
-import com.flipkart.varadhi.config.ServerConfig;
+import com.flipkart.varadhi.config.AppConfiguration;
 import com.flipkart.varadhi.spi.db.MetaStoreOptions;
 import com.flipkart.varadhi.spi.db.MetaStoreProvider;
 import com.flipkart.varadhi.spi.services.MessagingStackOptions;
@@ -39,7 +39,7 @@ public class CoreServices {
  private final MessagingStackProvider messagingStackProvider;
  private final MetaStoreProvider metaStoreProvider;
 
- public CoreServices(ServerConfig configuration) {
+ public CoreServices(AppConfiguration configuration) {
  this.observabilityStack = setupObservabilityStack(configuration);
  this.messagingStackProvider = setupMessagingStackProvider(configuration.getMessagingStackOptions());
  this.metaStoreProvider = setupMetaStoreProvider(configuration.getMetaStoreOptions());
@@ -50,7 +50,7 @@ public Tracer getTracer(String instrumentationScope) {
  return this.observabilityStack.getOpenTelemetry().getTracer(instrumentationScope);
  }
 
- public MeterRegistry getMetricsRegistry() {
+ public MeterRegistry getMeterRegistry() {
  return this.observabilityStack.getMeterRegistry();
  }
 
@@ -76,7 +76,8 @@ private MessagingStackProvider setupMessagingStackProvider(MessagingStackOptions
  return provider;
  }
 
- private ObservabilityStack setupObservabilityStack(ServerConfig configuration) {
+ private ObservabilityStack setupObservabilityStack(AppConfiguration configuration) {
+
  Resource resource = Resource.getDefault()
  .merge(Resource.create(Attributes.of(ResourceAttributes.SERVICE_NAME, "com.flipkart.varadhi")));
 

server/src/main/java/com/flipkart/varadhi/RestVerticle.java

@@ -102,8 +102,8 @@ public void start(Promise<Void> startPromise) {
  log.info("HttpServer Started.");
  startPromise.complete();
  } else {
- log.warn("HttpServer Start Failed.");
- startPromise.fail("HttpServer Start Failed.");
+ log.warn("HttpServer Start Failed." + h.cause());
+ startPromise.fail("HttpServer Start Failed." + h.cause());
  }
  });
  }