[release] lib-cmd-queue-redis@0.1.0

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: pditommaso

.github/workflows/build.yml

@@ -86,6 +86,7 @@ jobs:
           bash publish.sh lib-serde-jackson
           bash publish.sh lib-trace
           bash publish.sh lib-jedis-pool
+          bash publish.sh lib-cmd-queue-redis
           bash publish.sh micronaut-cache-redis
 
         env:

.github/workflows/generate-submit-dependencies.yml

@@ -37,6 +37,7 @@ jobs:
           "lib-trace",
           "lib-util-http",
           "lib-jedis-pool",
+          "lib-cmd-queue-redis",
           "micronaut-cache-redis",
           "wave-api",
           "wave-utils"

lib-cmd-queue-redis/README.md

@@ -0,0 +1,130 @@
+# lib-cmd-queue-redis
+
+Asynchronous command queue for executing long-running tasks with persistent state tracking and automatic status polling.
+
+## Installation
+
+Add this dependency to your `build.gradle`:
+
+```gradle
+dependencies {
+    implementation 'io.seqera:lib-cmd-queue-redis:0.1.0'
+}
+```
+
+## Features
+
+- Fire-and-forget command submission
+- Typed parameters and results with JSON serialization
+- Status transitions: `SUBMITTED` → `RUNNING` → `SUCCEEDED`/`FAILED`/`CANCELLED`
+- Automatic timeout handling for long-running commands
+- Periodic status checking for async commands
+- Command cancellation support
+- Persistent storage using Redis or in-memory backend
+
+## Usage
+
+### Define Command Parameters and Result
+
+```java
+// Command parameters - must have default constructor for Jackson
+public class ProcessingParams {
+    private String datasetId;
+    private List<String> steps;
+    // Getters, setters, constructors...
+}
+
+// Command result
+public class ProcessingResult {
+    private int recordsProcessed;
+    private long durationMs;
+    // Getters, setters, constructors...
+}
+```
+
+### Implement a Command Handler
+
+For **synchronous** commands that complete quickly:
+
+```java
+@Singleton
+public class ProcessingHandler implements CommandHandler<ProcessingParams, ProcessingResult> {
+
+    @Override
+    public String type() { return "data-processing"; }
+
+    @Override
+    public CommandResult<ProcessingResult> execute(Command<ProcessingParams> command) {
+        var params = command.params();
+        // Do the work...
+        var result = new ProcessingResult(1000, 5000L);
+        return CommandResult.success(result);
+    }
+}
+```
+
+For **asynchronous** long-running commands:
+
+```java
+@Singleton
+public class AsyncProcessingHandler implements CommandHandler<ProcessingParams, ProcessingResult> {
+
+    @Inject ExternalService externalService;
+
+    @Override
+    public String type() { return "async-processing"; }
+
+    @Override
+    public CommandResult<ProcessingResult> execute(Command<ProcessingParams> command) {
+        // Start async job
+        externalService.startJob(command.id(), command.params());
+        return CommandResult.running();  // checkStatus() will be called later
+    }
+
+    @Override
+    public CommandResult<ProcessingResult> checkStatus(Command<ProcessingParams> command, CommandState state) {
+        var status = externalService.getStatus(command.id());
+        if (status.isComplete()) return CommandResult.success(status.getResult());
+        if (status.isFailed()) return CommandResult.failure(status.getError());
+        return CommandResult.running();  // Still running, check again later
+    }
+}
+```
+
+### Submit Commands
+
+```java
+@Inject
+private CommandService commandService;
+
+// Register handler at startup
+commandService.registerHandler(new ProcessingHandler());
+
+// Submit a command
+var command = new ProcessingCommand("cmd-123", params);
+String commandId = commandService.submit(command);
+
+// Check status
+Optional<CommandState> state = commandService.getState(commandId);
+
+// Get result when complete
+ProcessingResult result = commandService.getResult(commandId, ProcessingResult.class).orElseThrow();
+```
+
+## Command Status Flow
+
+```
+submit() ──▶ SUBMITTED ──pickup──▶ RUNNING ─┬─success──▶ SUCCEEDED
+                                            ├─error────▶ FAILED
+                                            └─cancel───▶ CANCELLED
+```
+
+## Testing
+
+```bash
+./gradlew :lib-cmd-queue-redis:test
+```
+
+## License
+
+Apache License 2.0

lib-cmd-queue-redis/VERSION

@@ -0,0 +1 @@
+0.1.0

lib-cmd-queue-redis/build.gradle

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2025, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+plugins {
+    id 'io.seqera.java-library-conventions'
+    id 'groovy'
+}
+
+group = 'io.seqera'
+version = "${project.file('VERSION').text.trim()}"
+
+dependencies {
+    // Micronaut core
+    implementation "io.micronaut:micronaut-inject:${micronautCoreVersion}"
+    implementation "io.micronaut:micronaut-runtime:${micronautCoreVersion}"
+
+    // JSON serialization
+    implementation project(':lib-serde-jackson')
+
+    // Groovy (needed for AbstractMessageStream base class)
+    compileOnly "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
+
+    // Message stream
+    implementation project(':lib-data-stream-redis')
+    implementation project(':lib-serde-moshi')
+
+    // State store
+    implementation project(':lib-data-store-state-redis')
+
+    // Utilities (includes type resolution)
+    implementation project(':lib-lang')
+
+    // TSID for ID generation
+    implementation 'com.github.f4b6a3:tsid-creator:5.2.6'
+
+    // SLF4J logging
+    implementation 'org.slf4j:slf4j-api:2.0.17'
+
+    // Test dependencies
+    testImplementation "io.micronaut:micronaut-inject-groovy:${micronautCoreVersion}"
+    testImplementation "io.micronaut.test:micronaut-test-spock:${micronautTestVersion}"
+    testImplementation 'org.apache.groovy:groovy'
+    testImplementation 'com.github.f4b6a3:tsid-creator:5.2.6'
+    testImplementation 'org.junit.jupiter:junit-jupiter-api'
+    testRuntimeOnly 'ch.qos.logback:logback-classic:1.5.13'
+}

lib-cmd-queue-redis/changelog.txt

@@ -0,0 +1,12 @@
+# lib-cmd-queue-redis changelog
+
+0.1.0 - 15 Jan 2026
+- Initial release
+- Asynchronous command queue with persistent state tracking
+- Fire-and-forget command submission
+- Typed parameters and results with JSON serialization
+- Status transitions: SUBMITTED → RUNNING → SUCCEEDED/FAILED/CANCELLED
+- Automatic timeout handling for long-running commands
+- Periodic status checking via checkStatus() for async commands
+- Command cancellation support
+- Redis-backed persistent storage with TTL expiration

lib-cmd-queue-redis/src/main/java/io/seqera/data/command/Command.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2025, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.seqera.data.command;
+
+/**
+ * A command represents a unit of work to be executed asynchronously.
+ * Commands can take arbitrary time (from milliseconds to days) and survive system restarts.
+ *
+ * @param <P> The type of the command parameters
+ */
+public interface Command<P> {
+
+    /**
+     * Unique identifier for this command (typically a TSID).
+     */
+    String id();
+
+    /**
+     * The command type, used to route to the appropriate handler.
+     */
+    String type();
+
+    /**
+     * The typed parameters for this command.
+     */
+    P params();
+}

lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandConfig.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2025, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.seqera.data.command;
+
+import java.time.Duration;
+
+/**
+ * Configuration interface for command queue processing.
+ *
+ * <p>Provides default values for all configuration options. Applications
+ * can implement this interface to provide custom values via their
+ * preferred configuration mechanism (e.g., @Value annotations).
+ *
+ * @author Paolo Di Tommaso
+ */
+public interface CommandConfig {
+
+    /**
+     * Interval for polling the command queue.
+     */
+    default Duration pollInterval() {
+        return Duration.ofSeconds(1);
+    }
+
+    /**
+     * Timeout for synchronous command execution.
+     * If execute() takes longer than this, the command is marked as RUNNING
+     * and checkStatus() will be called on subsequent queue deliveries.
+     */
+    default Duration executeTimeout() {
+        return Duration.ofSeconds(1);
+    }
+
+    /**
+     * TTL (Time-To-Live) for command state records in the persistent store.
+     * Commands expire and are removed after this duration.
+     */
+    default Duration stateTtl() {
+        return Duration.ofDays(7);
+    }
+}

lib-cmd-queue-redis/src/main/java/io/seqera/data/command/CommandHandler.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2025, Seqera Labs
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package io.seqera.data.command;
+
+/**
+ * Handler for executing commands of a specific type.
+ * Generic types P (params) and R (result) are extracted via reflection at registration time.
+ *
+ * @param <P> The type of command parameters
+ * @param <R> The type of command result
+ */
+public interface CommandHandler<P, R> {
+
+    /**
+     * The command type this handler processes.
+     */
+    String type();
+
+    /**
+     * Execute the command and return a result.
+     * This method is executed asynchronously via an executor service.
+     * If execution takes longer than 1 second, the command is marked as RUNNING and
+     * {@link #checkStatus} will be called periodically to check completion.
+     * For long-running commands, return {@link CommandResult#running()} to indicate
+     * the operation is in progress.
+     *
+     * @param command The command to execute
+     * @return The result of the execution
+     */
+    CommandResult<R> execute(Command<P> command);
+
+    /**
+     * Check the status of a long-running command.
+     * Called periodically for commands in RUNNING state until a terminal status is returned.
+     * The command parameter provides typed access to params via {@code command.params()}.
+     * The state parameter provides access to timing and status information.
+     *
+     * @param command The command being checked (provides typed params access)
+     * @param state The current command state (timing, status info)
+     * @return The result indicating current status (RUNNING to continue, or terminal status)
+     */
+    default CommandResult<R> checkStatus(Command<P> command, CommandState state) {
+        return CommandResult.running();
+    }
+}