Vortex - Micro-Batching Library

import com.vajrapulse.vortex.*;
import com.vajrapulse.vortex.results.*;
import java.time.Duration;

// 1. Create a backend implementation
Backend<String> backend = batch -> {
    List<SuccessEvent<String>> successes = batch.stream()
        .map(SuccessEvent::new)
        .toList();
    return new BatchResult<>(successes, List.of());
};

// 2. Configure the batcher
BatcherConfig config = BatcherConfig.builder()
    .batchSize(10)                    // Batch up to 10 items
    .lingerTime(Duration.ofMillis(100)) // Or wait 100ms, whichever comes first
    .maxQueueSize(50)                 // Maximum 50 items in queue
    .build();

// 3. Create and use the batcher
try (MicroBatcher<String> batcher = new MicroBatcher<>(backend, config)) {
    // Submit with callback for async result handling
    ItemResult<String> result = batcher.submit("item-1", (item, itemResult) -> {
        if (itemResult instanceof ItemResult.Success<String>) {
            System.out.println("Item processed successfully: " + item);
        } else if (itemResult instanceof ItemResult.Failure<String> failure) {
            System.err.println("Item failed: " + failure.error().getMessage());
        }
    });
    
    // Check immediate acceptance/rejection
    if (result instanceof ItemResult.Success<String>) {
        System.out.println("Item accepted!");
    } else if (result instanceof ItemResult.Failure<String> failure) {
        System.err.println("Item rejected: " + failure.error().getMessage());
    }
}

// With callback (async result handling)
ItemResult<String> result = batcher.submit("item", itemResult -> {
    // Callback fires when item is processed (typically 10-50ms after submission)
    if (itemResult instanceof ItemResult.Success<String> success) {
        System.out.println("Success: " + success.getItem());
    } else if (itemResult instanceof ItemResult.Failure<String> failure) {
        System.err.println("Failed: " + failure.error().getMessage());
    }
});

// Without callback (fire and forget)
ItemResult<String> result = batcher.submit("item", null);

// Check immediate rejection
if (result instanceof ItemResult.Failure<String> failure) {
    // Item was rejected immediately (queue full, etc.)
    handleRejection(failure.error());
}

// Chain operations with CompletableFuture
CompletableFuture<ItemResult<String>> future = batcher.submitAsync("item");
future
    .thenApply(result -> {
        if (result instanceof ItemResult.Success<String>) {
            return processSuccess(result.getItem());
        } else if (result instanceof ItemResult.Failure<String> failure) {
            return handleFailure(failure.error());
        }
        return null;
    })
    .thenAccept(processed -> System.out.println("Processed: " + processed))
    .exceptionally(throwable -> {
        if (throwable instanceof ItemRejectedException) {
            // Handle immediate rejection
            handleRejection(throwable);
        }
        return null;
    });

// Or use with CompletableFuture.allOf() for batch operations
List<CompletableFuture<ItemResult<String>>> futures = items.stream()
    .map(batcher::submitAsync)
    .toList();
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
    .thenRun(() -> System.out.println("All items processed"));

// Success - item was accepted or processed successfully
ItemResult.Success<T> success = ItemResult.success(item);

// Failure - item was rejected or processing failed
ItemResult.Failure<T> failure = ItemResult.failure(item, error);

ItemResult<String> result = batcher.submit("item", null);

switch (result) {
    case ItemResult.Success<String> success -> 
        System.out.println("Success: " + success.item());
    case ItemResult.Failure<String> failure -> 
        System.err.println("Failed: " + failure.error().getMessage());
}

BatcherConfig config = BatcherConfig.builder()
    .batchSize(10)                           // Items per batch
    .lingerTime(Duration.ofMillis(100))      // Max wait time
    .maxQueueSize(50)                        // Queue capacity
    .build();

// High Throughput (large batches, longer wait)
BatcherConfig highThroughput = BatcherConfig.highThroughputPreset();
MicroBatcher<String> batcher = new MicroBatcher<>(backend, highThroughput, registry);

// Low Latency (small batches, short wait)
BatcherConfig lowLatency = BatcherConfig.lowLatencyPreset();
MicroBatcher<String> batcher = new MicroBatcher<>(backend, lowLatency, registry);

// Balanced (default)
BatcherConfig balanced = BatcherConfig.balancedPreset();
MicroBatcher<String> batcher = new MicroBatcher<>(backend, balanced, registry);

// Resilient (with retry)
Predicate<Throwable> retryable = e -> e instanceof IOException;
BatcherConfig resilient = BatcherConfig.resilientPreset(retryable);
MicroBatcher<String> batcher = new MicroBatcher<>(backend, resilient, registry);

BatcherConfig config = BatcherConfig.builder()
    // Batching
    .batchSize(20)                           // Batch up to 20 items
    .lingerTime(Duration.ofMillis(200))      // Or wait 200ms
    
    // Queue Management
    .maxQueueSize(100)                       // Max 100 items in queue
    .queueRejectionThreshold(0.9)            // Reject at 90% capacity
    
    // Concurrent Limiting
    .maxConcurrentBatches(5)                 // Max 5 concurrent batches
    
    // Retry
    .maxRetries(3)                           // Retry failed items up to 3 times
    .retryDelay(Duration.ofMillis(100))      // Wait 100ms between retries
    .retryableErrorPredicate(e ->            // Only retry transient errors
        e instanceof IOException || 
        e instanceof TimeoutException)
    
    // Atomic Commit
    .atomicCommit(true)                      // Batch fails if any item fails
    
    // Auto-Replay
    .autoReplaySuccesses(true)               // Replay successful items in mixed batches
    
    // Metrics
    .perItemMetrics(true)                    // Track per-item metrics
    
    // Debugging
    .debugMode(true)                         // Enable debug logging
    
    // Tracing
    .tracingHook(new LoggingTracingHook())  // Add tracing hook
    
    .build();

ItemResult<String> result = batcher.submit("item", null);

if (result instanceof ItemResult.Failure<String> failure) {
    Throwable error = failure.error();
    
    if (error instanceof ItemRejectedException rejected) {
        // Item was rejected
        System.out.println("Rejection source: " + rejected.getSourceName());
        System.out.println("Current level: " + rejected.getCurrentLevel());
        System.out.println("Max level: " + rejected.getMaxLevel());
        
        // Handle rejection
        handleRejection(rejected);
    }
}

BatcherConfig config = BatcherConfig.builder()
    .batchSize(10)                    // Items per batch
    .maxQueueSize(50)                 // Maximum items in queue
    .queueRejectionThreshold(0.8)     // Reject when queue is 80% full (0.0 to 1.0)
    .build();

ItemResult<String> result = batcher.submit("item", null);

if (result instanceof ItemResult.Failure<String> failure) {
    if (failure.error() instanceof ItemRejectedException rejected) {
        // Handle rejection: retry, overflow queue, fail fast, etc.
        handleRejection(item, rejected);
    }
}

BatcherConfig config = BatcherConfig.builder()
    .maxRetries(3)
    .retryDelay(Duration.ofMillis(100))
    .retryableErrorPredicate(e -> 
        e instanceof IOException || 
        e instanceof TimeoutException)
    .build();

// Failed items are automatically retried
batcher.submit("item", (item, result) -> {
    // This callback may be called multiple times if retries occur
    if (result instanceof ItemResult.Success<String>) {
        System.out.println("Item succeeded (possibly after retry)");
    }
});

BatcherConfig config = BatcherConfig.builder()
    .atomicCommit(true)  // Enable atomic commit
    .build();

// If any item fails, all items in the batch fail

try (MicroBatcher<String> batcher = new MicroBatcher<>(backend, config)) {
    // Submit items
    batcher.submit("item-1", null);
    batcher.submit("item-2", null);
    
    // Wait for all items to be processed
    boolean completed = batcher.awaitCompletion(5, TimeUnit.SECONDS);
    
    if (completed) {
        System.out.println("All items processed");
    } else {
        System.out.println("Timeout - some items may still be processing");
    }
    
    // close() is called automatically by try-with-resources
}

MetricsProvider metrics = batcher.getMetricsProvider();

// Rate metrics
double successRate = metrics.getSuccessRate();
double failureRate = metrics.getFailureRate();
double rejectionRate = metrics.getRejectionRate();

// Latency metrics
double avgLatency = metrics.getAverageLatency();
double p95Latency = metrics.getPercentileLatency(0.95);
double p99Latency = metrics.getPercentileLatency(0.99);

// Queue metrics
int queueDepth = batcher.getQueueDepth();

// ✅ Good - handles rejections
ItemResult<String> result = batcher.submit("item", null);
if (result instanceof ItemResult.Failure<String> failure) {
    handleRejection(failure.error());
}

// ❌ Bad - ignores rejections
batcher.submit("item", null);

// ✅ Good - automatic cleanup
try (MicroBatcher<String> batcher = new MicroBatcher<>(backend, config)) {
    // Use batcher
}

// ❌ Bad - manual cleanup (easy to forget)
MicroBatcher<String> batcher = new MicroBatcher<>(backend, config);
// ... use batcher ...
batcher.close(); // Easy to forget!

// ✅ Good - callback exceptions don't crash the system
batcher.submit("item", (item, result) -> {
    try {
        processResult(result);
    } catch (Exception e) {
        logger.error("Error processing result", e);
    }
});

// ✅ Good - use presets
BatcherConfig config = BatcherConfig.highThroughputPreset();

// ❌ Bad - manual configuration (may not be optimal)
BatcherConfig config = BatcherConfig.builder()
    .batchSize(100)
    .lingerTime(Duration.ofMillis(500))
    .maxQueueSize(500)
    .build();