feat: add etcd support for persistent context

build.gradle.kts

@@ -63,6 +63,8 @@ dependencies {
 
   implementation("com.google.protobuf:protobuf-java:4.32.0")
 
+  implementation("io.etcd:jetcd-core:0.8.5")
+
   implementation("io.nats:jnats:2.17.3")
 
   implementation(platform("io.opentelemetry:opentelemetry-bom:1.38.0"))

src/main/java/at/ac/uibk/dps/cirrina/cirrina/Cirrina.kt

@@ -1,7 +1,7 @@
 package at.ac.uibk.dps.cirrina.cirrina
 
 import at.ac.uibk.dps.cirrina.execution.`object`.context.Context
-import at.ac.uibk.dps.cirrina.execution.`object`.context.NatsContext
+import at.ac.uibk.dps.cirrina.execution.`object`.context.EtcdContext
 import at.ac.uibk.dps.cirrina.execution.`object`.event.EventHandler
 import at.ac.uibk.dps.cirrina.execution.`object`.event.NatsEventHandler
 import at.ac.uibk.dps.cirrina.execution.service.RandomServiceImplementationSelector
@@ -21,6 +21,7 @@ private val logger: FluentLogger = FluentLogger.forEnclosingClass()
 class Cirrina {
   companion object {
     const val NATS_CONNECTION_TIMEOUT = 60000L
+    const val ETCD_CONNECTION_TIMEOUT = 60000L
 
     init {
       ToStringBuilder.setDefaultStyle(SIMPLE_STYLE)
@@ -59,9 +60,9 @@ class Cirrina {
           logger.atFine().log("Creating the persistent context")
           newPersistentContext()
             .apply {
-              if (this is NatsContext) {
-                logger.atFine().log("Awaiting connection to NATS as the persistent context")
-                awaitInitialConnection(NATS_CONNECTION_TIMEOUT)
+              if (this is EtcdContext) {
+                logger.atFine().log("Awaiting connection to Etcd as the persistent context")
+                awaitInitialConnection(ETCD_CONNECTION_TIMEOUT)
               }
             }
             .use { persistentContext ->
@@ -116,21 +117,17 @@ class Cirrina {
   // Construct a new persistent context based as configured.
   private fun newPersistentContext(): Context =
     when (EnvironmentVariables.contextProvider.get()) {
-      PersistentContextProvider.NATS -> newNatsPersistentContext()
+      PersistentContextProvider.ETCD -> newEtcdPersistentContext()
       else ->
         throw ConfigurationError.Unknown(
           "persistent context",
           EnvironmentVariables.contextProvider.get(),
         )
     }
 
-  // Construct a new NATS persistent context as configured.
-  private fun newNatsPersistentContext(): NatsContext =
-    NatsContext(
-      false,
-      EnvironmentVariables.natsContextUrl.get(),
-      EnvironmentVariables.natsContextBucket.get(),
-    )
+  // Construct a new Etcd persistent context as configured.
+  private fun newEtcdPersistentContext(): EtcdContext =
+    EtcdContext(false, listOf(EnvironmentVariables.etcdContextUrl.get()))
 
   // Construct a new OpenTelemetry instance as configured.
   private fun getOpenTelemetry(): OpenTelemetry =

src/main/java/at/ac/uibk/dps/cirrina/cirrina/EnvironmentVariables.kt

@@ -7,7 +7,7 @@ enum class EventProvider {
 
 /** Provider for the persistent context. */
 enum class PersistentContextProvider {
-  NATS
+  ETCD
 }
 
 /** An environment variable, which can be converted from a string to the required type. */
@@ -38,11 +38,8 @@ object EnvironmentVariables {
   /** The NATS event server URL. */
   val natsEventUrl = EnvironmentVariable("NATS_EVENT_URL", default = "nats://localhost:4222/")
 
-  /** The NATS context server URL. */
-  val natsContextUrl = EnvironmentVariable("NATS_CONTEXT_URL", default = "nats://localhost:4222")
-
-  /** The NATS context bucket. */
-  val natsContextBucket = EnvironmentVariable("NATS_CONTEXT_BUCKET", default = "persistent")
+  /** The Etcd context server URL. */
+  val etcdContextUrl = EnvironmentVariable("ETCD_CONTEXT_URL", default = "http://localhost:2379")
 
   /** The path to the CSML application. */
   val appPath = EnvironmentVariable<String>("APP_PATH", required = true)
@@ -80,7 +77,7 @@ object EnvironmentVariables {
   val contextProvider =
     EnvironmentVariable(
       name = "CONTEXT_PROVIDER",
-      default = PersistentContextProvider.NATS,
+      default = PersistentContextProvider.ETCD,
       mapper = { value ->
         try {
           PersistentContextProvider.valueOf(value.uppercase())

src/main/java/at/ac/uibk/dps/cirrina/execution/object/context/Context.java

@@ -13,7 +13,7 @@ public abstract class Context implements AutoCloseable {
   /**
    * Initializes this context object.
    *
-   * @param isLocal True if this context is local, otherwise false.
+   * @param isLocal true if this context is local, otherwise false
    */
   public Context(boolean isLocal) {
     this.isLocal = isLocal;
@@ -22,29 +22,29 @@ public Context(boolean isLocal) {
   /**
    * Retrieve a context variable.
    *
-   * @param name Name of the context variable.
-   * @return The retrieved context variable.
-   * @throws IOException If the context variable could not be retrieved.
+   * @param name name of the context variable
+   * @return The retrieved context variable
+   * @throws IOException if the context variable could not be retrieved
    */
   public abstract Object get(String name) throws IOException;
 
   /**
    * Creates a context variable.
    *
-   * @param name  Name of the context variable.
-   * @param value Value of the context variable.
-   * @return Byte size of stored data.
-   * @throws IOException If the variable could not be created.
+   * @param name  name of the context variable
+   * @param value value of the context variable
+   * @return byte size of stored data
+   * @throws IOException if the variable could not be created
    */
   public abstract int create(String name, Object value) throws IOException;
 
   /**
    * Assigns to a context variable.
    *
-   * @param name  Name of the context variable.
-   * @param value New value of the context variable.
-   * @return Byte size of stored data.
-   * @throws IOException If the variable could not be assigned to.
+   * @param name  name of the context variable
+   * @param value new value of the context variable
+   * @return byte size of stored data
+   * @throws IOException if the variable could not be assigned to
    */
   public abstract int assign(String name, Object value) throws IOException;
 
@@ -56,18 +56,25 @@ public Context(boolean isLocal) {
    */
   public abstract void delete(String name) throws IOException;
 
+  /**
+   * Deletes all context variables.
+   *
+   * @throws IOException if the variable could not be deleted
+   */
+  public abstract void deleteAll() throws IOException;
+
   /**
    * Returns all context variables.
    *
-   * @return Context variables.
-   * @throws IOException If the variables could not be retrieved.
+   * @return context variables.
+   * @throws IOException if the variables could not be retrieved
    */
   public abstract List<ContextVariable> getAll() throws IOException;
 
   /**
    * Returns a flag that indicates if this context is local.
    *
-   * @return True if local, otherwise false.
+   * @return true if local, otherwise false
    */
   public boolean isLocal() {
     return isLocal;

src/main/java/at/ac/uibk/dps/cirrina/execution/object/context/ContextBuilder.java

@@ -61,23 +61,6 @@ public ContextBuilder inMemoryContext(boolean isLocal) {
     return this;
   }
 
-  /**
-   * Build a NATS context.
-   *
-   * @param isLocal    True if this context is local, otherwise false.
-   * @param natsUrl    NATS url.
-   * @param bucketName NATS bucket name.
-   * @return This builder.
-   * @throws IOException If the context could not be built.
-   * @see NatsContext
-   */
-  public ContextBuilder natsContext(boolean isLocal, String natsUrl, String bucketName)
-    throws IOException {
-    context = new NatsContext(isLocal, natsUrl, bucketName, false);
-
-    return this;
-  }
-
   /**
    * Builds the current context.
    *