Functional Interfaces

Ratchet defines six serializable functional interfaces for job tasks, callbacks, and workflow conditions. All extend both their java.util.function counterpart (where applicable) and Serializable, so Ratchet can analyze and persist job definitions in the job store.

Package: run.ratchet.api

Why serializable?

Ratchet persists job definitions in a database. When you write:

scheduler.enqueue(() -> orderService.process(orderId)).submit();

The lambda () -> orderService.process(orderId) must be serializable so Ratchet can inspect it, store a portable payload, and execute it later on another node. All functional interfaces in Ratchet extend Serializable to make this possible.

Method Reference Constraint

Job tasks and workflow condition predicates must contain exactly one method invocation (a single method reference or method call). Multi-statement lambdas and inline boolean logic fail at submission time with IllegalArgumentException. This constraint applies to task-defining interfaces and to condition interfaces used by when(), whenResult(), and thenWhenBatch().

For complex logic, create a dedicated method in a CDI bean and reference it.

SerializableCheckedRunnable

The primary functional interface for defining job tasks. Extends Serializable and allows checked exceptions.

@FunctionalInterface
public interface SerializableCheckedRunnable extends Serializable {
    void run() throws Exception;
}

Used in: JobSchedulerService.enqueue(), schedule(), enqueueNow(), JobBuilder.then(), all workflow branch tasks.

Correct usage

// Method reference
scheduler.enqueue(myService::processData).submit();

// Instance method reference
scheduler.enqueue(myService::sendEmail).submit();

// Single method call with parameters
scheduler.enqueue(() -> myService.process(userId)).submit();

// Single method call with multiple parameters
scheduler.enqueue(() -> reportService.generate(userId, reportType, startDate)).submit();

Incorrect usage

// WRONG -- multi-statement lambda
scheduler.enqueue(() -> {
    processData();
    updateDatabase();
}).submit();  // throws IllegalArgumentException

// WRONG -- multiple invocations
scheduler.enqueue(() -> {
    User user = userService.findById(userId);
    notificationService.send(user, message);
}).submit();  // throws IllegalArgumentException

Workaround for complex logic

Wrap multi-step logic in a single method on a CDI bean:

@ApplicationScoped
public class UserWorkflow {
    @Inject UserService userService;
    @Inject NotificationService notifications;

    public void processAndNotify(String userId) {
        User user = userService.findById(userId);
        userService.processData(user);
        notifications.send(user, "Processing complete");
    }
}

// Single method reference
scheduler.enqueue(() -> userWorkflow.processAndNotify(userId)).submit();

Exception handling

Exceptions thrown from run() are caught by the executor and trigger:

1. Retry attempts based on job configuration
2. Backoff delays between retries
3. Dead letter queue processing after max retries exhausted
4. Failure callbacks if configured via onFailure()

SerializableConsumer<T>

Serializable variant of java.util.function.Consumer<T>. Accepts a single argument and returns no result.

@FunctionalInterface
public interface SerializableConsumer<T> extends Consumer<T>, Serializable {
    void accept(T t);
}

Used in: JobBuilder.onSuccess() (as SerializableConsumer<JobContext>), BatchBuilder.onProgress() (as SerializableConsumer<BatchContext>), BatchBuilder.forEach().

Examples

// Success callback
scheduler.enqueue(() -> processOrder(orderId))
    .onSuccess(ctx -> log.info("Job {} succeeded", ctx.jobId()))
    .submit();

// Batch progress monitoring
scheduler.enqueueBatch("Import")
    .forEach(records, record -> importRecord(record))
    .onProgress(ctx -> log.info("{}% complete", ctx.percentDone()))
    .submit();

// Method reference
scheduler.enqueue(() -> processOrder(orderId))
    .onSuccess(this::handleSuccess)
    .submit();

private void handleSuccess(JobContext ctx) {
    updateOrderStatus(ctx.param("orderId"), "COMPLETE");
}

SerializableBiConsumer<T, U>

Serializable variant of java.util.function.BiConsumer<T, U>. Accepts two arguments and returns no result.

@FunctionalInterface
public interface SerializableBiConsumer<T, U> extends BiConsumer<T, U>, Serializable {
    void accept(T t, U u);
}

Used in: JobBuilder.onFailure() (as SerializableBiConsumer<JobContext, Throwable>).

Examples

// Failure callback with context and error
scheduler.enqueue(() -> riskyOperation())
    .onFailure((ctx, error) -> {
        log.error("Job {} failed: {}", ctx.jobId(), error.getMessage());
        alertService.sendAlert(ctx.jobId(), error);
    })
    .submit();

// Method reference
scheduler.enqueue(() -> riskyOperation())
    .onFailure(this::handleFailure)
    .submit();

private void handleFailure(JobContext ctx, Throwable error) {
    log.error("Job {} failed", ctx.jobId(), error);
}

SerializableFunction<T, R>

Serializable variant of java.util.function.Function<T, R>. Accepts one argument and produces a result.

@FunctionalInterface
public interface SerializableFunction<T, R> extends Function<T, R>, Serializable {
    R apply(T t);
}

Used in: JobBuilder.whenResult() (as SerializableFunction<T, Boolean>), WorkflowCondition.result().

Examples

public final class ScoreConditions {
    public static boolean isHighPriority(Double score) {
        return score > 0.8;
    }

    public static boolean isLowPriority(Double score) {
        return score <= 0.5;
    }

    public static boolean hasGoldSubscription(UserData user) {
        return user.getSubscription() == Premium.GOLD;
    }
}

// Value-based workflow branching
scheduler.enqueue(() -> analyzeData())
    .whenResult(ScoreConditions::isHighPriority, () -> triggerHighPriority())
    .whenResult(ScoreConditions::isLowPriority, () -> triggerLowPriority())
    .submit();

// Complex value-based condition
scheduler.enqueue(() -> fetchUserData())
    .whenResult(ScoreConditions::hasGoldSubscription,
                () -> sendPremiumFeatures())
    .submit();

// Method reference for conditions
scheduler.enqueue(() -> checkInventory())
    .whenResult(this::isLowStock, () -> reorder())
    .submit();

public Boolean isLowStock(Integer stockLevel) {
    return stockLevel < 10;
}

SerializablePredicate<T>

Serializable variant of java.util.function.Predicate<T>. Accepts one argument and returns a boolean.

@FunctionalInterface
public interface SerializablePredicate<T> extends Predicate<T>, Serializable {
    boolean test(T t);
}

Used in: JobBuilder.when() (as SerializablePredicate<JobResult<T>>), BatchBuilder.thenWhenBatch() (as SerializablePredicate<BatchContext>), WorkflowCondition.custom(), WorkflowCondition.batchCustom().

Examples

public final class WorkflowPredicates {
    public static boolean isFastSuccess(JobResult<?> result) {
        return result.isSuccess()
            && result.getExecutionTimeMsOrZero() < 5_000;
    }

    public static boolean isTimeoutFailure(JobResult<?> result) {
        String error = result.getError();
        return result.isFailure()
            && error != null
            && error.contains("timeout");
    }

    public static boolean allSucceeded(BatchContext ctx) {
        return ctx.failedItems() == 0;
    }

    public static boolean needsRollback(BatchContext ctx) {
        return ctx.successRate() < 0.5;
    }

    public static boolean isCriticalBatch(BatchContext ctx) {
        return ctx.failedItems() > 10 || ctx.successRate() < 0.5;
    }
}

// Job result condition
scheduler.enqueue(() -> processData())
    .when(WorkflowPredicates::isFastSuccess,
          () -> log.info("Fast execution"))
    .when(WorkflowPredicates::isTimeoutFailure,
          () -> increaseTimeoutAndRetry())
    .submit();

// Batch context condition
scheduler.enqueueBatch("Import")
    .forEach(records, r -> importRecord(r))
    .thenWhenBatch(WorkflowPredicates::allSucceeded, () -> markComplete())
    .thenWhenBatch(WorkflowPredicates::needsRollback, () -> rollback())
    .submit();

// Combining conditions
SerializablePredicate<BatchContext> criticalCondition =
    WorkflowPredicates::isCriticalBatch;

scheduler.enqueueBatch("Critical Process")
    .forEach(items, item -> processItem(item))
    .thenWhenBatch(criticalCondition, () -> escalateToOps())
    .submit();

SerializableCheckedConsumer<T>

Serializable consumer that can throw checked exceptions. Unlike SerializableConsumer, this does not extend java.util.function.Consumer because Consumer does not support checked exceptions.

@FunctionalInterface
public interface SerializableCheckedConsumer<T> extends Serializable {
    void accept(T t) throws Exception;
}

Used in: StreamingBatchBuilder.process() for streaming batch item processing where the action may throw checked exceptions (e.g., SQLException, IOException).

Examples

// Streaming batch with checked exception handling
scheduler.<Long>streamingBatch("Process Users")
    .fromStream(userRepository.streamAllUserIds())
    .process(userId -> {
        // This can throw checked exceptions
        userService.processUser(userId);
    })
    .start();

// Method reference
scheduler.<Long>streamingBatch("Migrate Users")
    .fromStream(userIds.stream())
    .process(migrationService::migrateUser)
    .start();

Comparison with SerializableConsumer

Feature	SerializableConsumer	SerializableCheckedConsumer
Extends	Consumer<T> + Serializable	Serializable only
Checked exceptions	No	Yes
Primary use	Callbacks, progress hooks	Streaming batch processing

See also

• JobBuilder Reference
• BatchBuilder Reference
• WorkflowCondition Reference