OxMQThe Virtual Thread-Native Distributed Queue for Java 17+ & 21

// Cooperative AbortSignal parity on Java 21 Virtual Threads
OxmqWorker<VideoTask> worker = OxmqWorker.<VideoTask>builder()
        .queueName("video-rendering")
        .redisUri("redis://localhost:6379")
        .processor(job -> {
            // Listen to cooperative cancellation token
            job.getCancellationToken().onCancel(reason -> cleanupTempFiles(job.getId()));
            
            while (!job.isCancelled()) {
                renderNextFrame();
            }
            throw new JobCancelledException(job.getCancelReason());
        })
        .build();
worker.start();

// Externally cancel long-running job by ID
worker.cancelJob("video-job-42", "User clicked abort");

// Dequeue up to 100 jobs atomically with maxBatchBytes memory guardrail
OxmqBatchWorker<AuditRecord> batchWorker = OxmqBatchWorker.<AuditRecord>builder()
        .queueName("audit-stream")
        .redisUri("redis://localhost:6379")
        .batchSize(100)
        .maxBatchBytes(16 * 1024 * 1024) // 16 MB payload memory safeguard
        .batchTimeout(Duration.ofMillis(200))
        .processor(batch -> {
            List<AuditRecord> records = batch.stream().map(Job::getData).toList();
            clickHouseService.batchInsert(records); // Single bulk JDBC transaction!
            return "INSERTED_" + records.size();
        })
        .build();
batchWorker.start();

@Component
public class PaymentWorker {

    // Automatically runs on lightweight Virtual Threads in Java 21+
    @OxmqListener(queue = "orders", concurrency = 50, rateLimitMax = 100, rateLimitDurationMs = 1000)
    public PaymentResult handlePayment(Job<OrderPayload> job) {
        job.updateProgress(25);
        job.log("Charging credit card for order " + job.getData().orderId());
        return paymentGateway.charge(job.getData());
    }
}

@RestController
@RequestMapping("/api/orders")
public class OrderController {

    @Autowired
    private OxmqTemplate oxmqTemplate;

    @PostMapping("/checkout")
    public CompletableFuture<ResponseEntity<PaymentResult>> checkout(@RequestBody OrderPayload order) {
        Job<OrderPayload> job = oxmqTemplate.send("orders", "charge-card", order);
        
        // Reactively await completion via Redis stream events without blocking carrier threads
        return oxmqTemplate.waitUntilFinishedAsync("orders", job.getId(), Duration.ofSeconds(15))
                .thenApply(res -> ResponseEntity.ok((PaymentResult) res))
                .exceptionally(ex -> ResponseEntity.status(HttpStatus.GATEWAY_TIMEOUT).build());
    }
}

@Configuration
public class OxmqEnterprisePluginConfig {

    // 1. Dynamic Token Bucket per Tenant / Group Key
    @Bean
    public GroupRateLimiterPlugin groupRateLimiter(RedisConnectionManager conn, MeterRegistry meter) {
        return GroupRateLimiterPlugin.builder()
                .connectionManager(conn)
                .keyPrefix("oxmq:limiter:tenant")
                .groupKeyResolver(job -> ((TenantScopedPayload) job.getData()).tenantId())
                .dynamicLimit(tenantId -> licenseService.getBurstLimit(tenantId)) // e.g. 50-200 req/sec
                .windowDuration(Duration.ofSeconds(1))
                .retryDelayMs(150)
                .meterRegistry(meter)
                .build();
    }

    // 2. Upstream 429/503 Circuit Breaker Sentinel with Non-Destructive Delay
    @Bean
    public RateLimitSentinelPlugin rateLimitSentinel(RedisConnectionManager conn, MeterRegistry meter) {
        return RateLimitSentinelPlugin.builder()
                .connectionManager(conn)
                .keyPrefix("oxmq:sentinel:tenant")
                .groupKeyResolver(job -> ((TenantScopedPayload) job.getData()).tenantId())
                .errorMatcher(t -> t instanceof RateLimitCooldownException)
                .meterRegistry(meter)
                .build();
    }

    // 3. Priority Headroom Governor: Protect real-time jobs by yielding background sweeps
    @Bean
    public PriorityQuotaGovernorPlugin quotaGovernor(GroupRateLimiterPlugin limiter, MeterRegistry meter) {
        return PriorityQuotaGovernorPlugin.builder()
                .rateLimiter(limiter)
                .groupKeyResolver(job -> ((TenantScopedPayload) job.getData()).tenantId())
                .headroomYieldThreshold(0.25) // If < 25% token headroom remains
                .priorityThreshold(1)         // Strictly protect Priority <= 1 real-time jobs
                .yieldDelayMs(200)
                .meterRegistry(meter)
                .build();
    }
}