Reactive events and streams

This guide shows how to follow a JOpt optimization run through its event streams: progress, status, warnings, errors, node filtering events and intermediate results. It is for Java developers who log, monitor or display a run, or who connect it to a reactive framework such as Spring WebFlux. It assumes an optimization in Java as built in the First optimization tutorial.

Table of contents

• What it does
  • Event subjects
• When to use it
• How to use it
  • Subscribing to events
  • On-demand progress requests
  • Structured progress data
  • Error handling with RxJava
  • Uncaught exception handling
  • Node filtering events
  • Legacy callbacks backed by subscriptions
  • Combining and coupling observables
  • Recommended integration patterns
  • Integration with Spring WebFlux
• Properties
• Limits
• Related examples
• Related pages

What it does

JOpt.TourOptimizer is built on a reactive, stream-based event architecture. Every signal that the optimizer produces during a run (progress updates, warnings, status changes, errors, node filtering events, intermediate results) is emitted through RxJava ReplaySubjects. Any number of consumers can subscribe to these subjects, for real-time monitoring, structured logging, UI updates and the integration with reactive service frameworks such as Spring WebFlux.

The event system is part of the core architecture. The same reactive pipeline that drives the internal optimization phases is open to the integrator. You subscribe to the events you need, process them without blocking the optimizer, and connect them to the event-driven parts of your application.

Event subjects

The optimizer exposes its events through getOptimizationEvents(), which returns an OptimizationEvents instance. This object gives access to several typed ReplaySubjects and to a CompletableFuture for the final result:

All ReplaySubjects complete automatically when the result future completes, normally or exceptionally, so their resources are released however the optimization ends.

Each subject is independent: you can subscribe to any combination of them, and several subscribers can attach to the same subject without interfering.

A late subscription is safe. A subscriber that attaches after the optimization has started receives the most recent buffered events, so a UI component that connects mid-run sees the current progress at once instead of an empty state.

When to use it

Use the event streams when you watch a run while it runs or react to it:

• progress bars and dashboards that show the phase, the progress and the cost of a run;
• logging and telemetry, with the events forwarded into your own pipeline;
• audit reports of the nodes that AutoFilter removed, with their reasons;
• stage-aware checks in tests and validation, with RxJava operators;
• REST services on Spring WebFlux that stream the progress to their clients.

How to use it

Subscribing to events

Subscribing takes one statement per event type. The subscription receives a lambda that processes each emitted event:

opti.getOptimizationEvents()
    .progressSubject()
    .subscribe(p -> System.out.println(p.getProgressString()));

opti.getOptimizationEvents()
    .warningSubject()
    .subscribe(w -> System.out.println(w.toString()));

opti.getOptimizationEvents()
    .statusSubject()
    .subscribe(s -> System.out.println(s.getDescription() + " " + s.getCode()));

opti.getOptimizationEvents()
    .errorSubject()
    .subscribe(e -> System.out.println(e.getCause() + " " + e.getCode()));

You can create several subscriptions to one event source. For example, one subscription logs to stdout while another pushes to a metrics pipeline; both receive the same events independently.

On-demand progress requests

Besides the automatic progress events, JOpt supports progress requests: call requestProgress() at any time to make the optimizer emit its current state, regardless of the configured frequency.

This is useful for:

• external polling intervals, for example a REST endpoint that returns the current status every 5 seconds;
• manual checkpoints in batch workflows;
• external monitoring systems that pull rather than subscribe.

// Request current progress from an external thread
opti.requestProgress();

The requested progress is emitted through the same progressSubject(), so your existing subscribers receive it without more wiring.

Structured progress data

Each progress event carries an IOptimizationProgress object with structured data:

• the current cost and improvement metrics;
• the algorithm phase (SA, Genetic, Construction);
• the progress percentage within the current phase;
• the current best (winner) solution;
• a human-readable progress string for quick logging.

With this data you can build:

• UI progress bars that show the phase;
• stage-based early stopping (a sample rule: "if the genetic evolution phase is at 95% and the cost has not improved in 5 minutes, stop");
• telemetry dashboards that track the cost convergence over time;
• automated alerts when the optimization quality stops improving.

For dashboards, sample the structured progress at a fixed wall-clock interval (for example every 10 seconds) rather than on every event, and store snapshots of the cost, distance and time metrics for convergence charts.

Error handling with RxJava

RxJava has error propagation rules that matter in production. If the onNext handler of a subscriber throws an exception, RxJava routes it through the error channel. Without an explicit error consumer, the exception becomes an OnErrorNotImplementedException and may reach the global RxJava error handler, with confusing stack traces.

Always provide an error consumer when you subscribe:

// The error consumer hands an exception of the subscriber to the uncaught exception handler of
// the setup, the same handler that JOpt uses for its own threads.
Consumer<? super Throwable> errorConsumer =
    t -> opti.getOptimizationSetup()
        .getUncaughtExceptionHandler()
        .uncaughtException(Thread.currentThread(), t);

// Subscribe with the error consumer
opti.getOptimizationEvents()
    .progressSubject()
    .subscribe(
        p -> processProgress(p),  // onNext
        errorConsumer              // onError
    );

The error consumer hands subscription failures to the uncaught exception handler of the setup, which calls your handler. JOpt does not end the optimization by itself: when the handler completes the result future exceptionally, as the example handler does, the wait for the result ends with the exception as its cause.

Uncaught exception handling

JOpt can attach a custom UncaughtExceptionHandler to the optimization, through its setup. JOpt wraps it in its own handler: when an exception in an optimizer thread is not caught by a normal try/catch block, JOpt's handler logs it, streams the error event UNCAUGHT_EXCEPTION_IN_OPTIMIZATION and then calls your handler; it does not terminate the optimization itself.

The recommended pattern sets the handler on the IOptimizationSetup and passes the setup to the constructor of the optimization (Optimization(IOptimizationSetup)):

IOptimizationSetup setup = new DefaultOptimizationSetup();
// JOpt wraps the handler in its own handler, which logs the exception, streams the error event
// UNCAUGHT_EXCEPTION_IN_OPTIMIZATION and then calls MyUncaughtExceptionHandler.
setup.setSlaveUncaughtExceptionHandler(new MyUncaughtExceptionHandler());

For a handler set this way, JOpt attaches the optimization before each call; a handler that you call yourself needs attachOptimization(opti).

When called, the handler should:

1. log the error with context (optimization ID, thread name, algorithm phase);
2. complete the result future exceptionally, as the example handler does with opti.getOptimizationEvents().resultFuture().completeExceptionally(e);
3. avoid heavy blocking IO inside the handler, to keep it short and deterministic.

This gives a clean shutdown path: the calling code handles the exception through the error chain of the CompletableFuture (exceptionally(...), handle(...)) instead of dealing with orphaned threads or silent failures.

Node filtering events

Node filtering is reactive too. nodeFilteringSubject() emits NodeFilteringEvent objects whenever AutoFilter excludes a node. Two snapshot subjects emit the optimization result right before and right after filtering, which shows what changed:

// Subscribe to individual filtering events
opti.getOptimizationEvents()
    .nodeFilteringSubject()
    .subscribe(ev -> {
        System.out.println("Filtered: " + ev.getCode().getCodeMessage());
        ev.getFilterReasons().forEach(reason ->
            System.out.println("  Reason: " + reason));
    });

// Subscribe to the result snapshot before filtering
opti.getOptimizationEvents()
    .beforeNodeFilteringResultSubject()
    .subscribe(result -> analyzePreFilterState(result));

// Subscribe to the result snapshot after filtering
opti.getOptimizationEvents()
    .afterNodeFilteringResultSubject()
    .subscribe(result -> analyzePostFilterState(result));

The before and after result subjects show what the schedule looked like before AutoFilter intervened. That supports audit reports: the plan before the exclusion, the plan after it, and the removed nodes with their reasons.

Legacy callbacks backed by subscriptions

JOpt also supports a callback interface (IOptimizationEventListener) with methods such as onProgress(...), onError(...), onWarning(...), onStatus(...) and onNodeFiltering(...). These callbacks use the same event system: internally, they are subscriptions to the same ReplaySubjects, made through a LegacyCallbackSubscriber.

When you override onProgress(IOptimizationProgress progress) in your Optimization subclass, you subscribe a lambda to progressSubject(). The same holds for the other callback methods. So:

• Callbacks and direct subscriptions coexist safely; you can use both in the same application.
• The error handling rules of RxJava apply to callbacks too. If your onProgress(...) override throws an exception, it becomes a subscription error. Attach an error consumer through the uncaught exception handler even when you use callbacks.
• Legacy callbacks receive both the typed event object and the convenience string or code. For example, onProgress is called once with the progress string and once with the structured IOptimizationProgress object.

For new integrations, subscribe to the subjects directly: that gives you full control over error consumers, threading and composition. The callback interface stays available for simple cases and for backward compatibility.

Combining and coupling observables

All event subjects are standard RxJava Observables, so the RxJava operators can combine them. This helps where you correlate information from several event streams.

Combining results with progress

A common pattern couples the requested result stream with the progress stream, for example to run validation checks at specific progress percentages:

Observable<IOptimizationResult> resultsOb =
    opti.getOptimizationEvents().requestedresult;
Observable<IOptimizationProgress> progressOb =
    opti.getOptimizationEvents().progress;

// Combine: for each new result, pair it with the latest progress
Observable<Boolean> validationResults =
    resultsOb.withLatestFrom(progressOb, (result, progress) -> {

        if (progress.getProgress() == 50.0) {
            // At 50%, check that no route has overtime
            return result.getRoutes().stream()
                .allMatch(r -> r.getRouteCostAndViolationController()
                    .getViolationSummary()
                    .getNumViolationsRouteTime() == 0);
        }

        if (progress.getProgress() == 95.0) {
            // At 95%, check that all routes are finalized
            return result.getRoutes().stream()
                .allMatch(r -> r.isFinalized());
        }

        return true;
    });

validationResults.subscribe(passed -> {
    if (!passed) {
        System.out.println("Validation failed at current progress");
    }
});

The withLatestFrom operator pairs each emitted result with the most recent progress event. That gives stage-aware validation without polling or manual state tracking.

Filtering subjects with predicates

The utility method createBoundSubject creates a derived subject that emits only the events that match a predicate:

// Create a subject that only emits progress events from the genetic stage
ReplaySubject<IOptimizationProgress> geneticProgress =
    OptimizationEvents.createBoundSubject(
        opti.getOptimizationEvents().progressSubject(),
        p -> p.getCallerId().contains("Genetic"),
        10  // buffer size
    );

geneticProgress.subscribe(p ->
    System.out.println("Genetic stage: " + p.getProgressString()));

Use it for stage-specific dashboards, for actions only in certain algorithm phases, or to route different event subsets to different consumers.

Testing with TestObserver

In integration tests, an RxJava TestObserver can be attached to any subject or derived observable, with assertions directly on the event stream:

TestObserver<OptimizationEvents.AssertionResult> testObserver =
    OptimizationEvents.subscribeSubjectWithTestObserver(
        opti.getOptimizationEvents().progressSubject(),
        p -> new OptimizationEvents.AssertionResult(
            p.getProgress() >= 0 && p.getProgress() <= 100,
            "Progress must be between 0 and 100"
        )
    );

// After optimization completes:
testObserver.assertNoErrors();
testObserver.values().forEach(ar -> assertTrue(ar.passed(), ar.message()));

Recommended integration patterns

Pattern 1: synchronous with callbacks

This is the simplest pattern. Start synchronously and override the methods onProgress, onError and onWarning directly. It suits CLI tools, batch jobs and quick prototypes.

Pattern 2: asynchronous with a CompletableFuture

This pattern suits services. Start with startRunAsync() and chain the result processing with thenAccept, handle or whenComplete, so that request threads do not block.

Pattern 3: reactive with event subscriptions

This pattern is recommended for applications that must be observable. Start asynchronously, subscribe to the event subjects for continuous visibility, and use the CompletableFuture for the final result:

// 1. Build model
addNodes(opti);
addResources(opti);

// 2. Subscribe to events
opti.getOptimizationEvents().progressSubject()
    .subscribe(p -> metricsService.recordProgress(p));
opti.getOptimizationEvents().errorSubject()
    .subscribe(e -> alertService.raiseError(e));

// 3. Start async
CompletableFuture<IOptimizationResult> future = opti.startRunAsync();

// 4. Handle result
future.thenAccept(result -> resultService.publish(result))
      .exceptionally(ex -> { errorService.handle(ex); return null; });

Integration with Spring WebFlux

The reactive event architecture maps onto Spring WebFlux and other reactive frameworks. Because JOpt emits its events through RxJava subjects, they can be bridged to Project Reactor (Flux and Mono) with little wiring.

The TourOptimizer REST service is a Spring WebFlux application with non-blocking HTTP handling. This means:

• Progress streaming. Server-Sent Events (SSE) or WebSocket endpoints can forward progress events from the optimizer to the client in real time. The RxJava subject is the source; a thin adapter converts it to a Reactor Flux.
• Non-blocking job control. Optimization runs start asynchronously. The CompletableFuture result fits the asynchronous response model of WebFlux, which avoids thread starvation under load.
• Scalable concurrency. Several optimization jobs can run at the same time, each with its own event stream. WebFlux handles the concurrent HTTP connections without blocking threads.
• Error propagation. Exceptions in optimizer threads propagate through the result future and can be mapped to HTTP error responses.

The REST endpoint of JOpt.TourOptimizer (Docker-REST-TourOptimizer) uses this architecture. Its Swagger UI at localhost:8081/swagger-ui/index.html shows the full optimization API, with reactive request and response handling.

For Python, C#, and Angular clients, the REST interface abstracts the reactive internals. These clients interact through standard HTTP requests and receive results as JSON. The reactive architecture is an internal implementation detail that benefits all clients through better throughput and lower latency.

Properties

Progress events can fire very often during a run. Emitting too often slows the optimizer (logging overhead) and floods the consumers downstream; emitting too rarely makes a UI feel unresponsive. Two properties set the progress frequency per stage; the values in this sample are examples:

// SA stage: emit progress every 0.1%
props.setProperty("JOpt.Algorithm.PreOptimization.SA.OnProgressOutPercentage", "0.1");

// Genetic stage: emit progress every 0.01%
props.setProperty("JOpt.Algorithm.GE.OnProgressOutPercentage", "0.01");

The simulated annealing stage is typically iteration-heavy and benefits from a moderate output rate. The genetic evolution phase runs longer and may justify more frequent updates for responsive UIs. The defaults of JOpt.Algorithm.PreOptimization.SA.OnProgressOutPercentage and JOpt.Algorithm.GE.OnProgressOutPercentage are on Optimization properties.

Limits

The event streams are RxJava ReplaySubjects with size-bounded buffers. That has these consequences:

Buffering is bounded. The internal buffers are sized per subject type: progress buffers 10 events, error/warning/status buffer 20 each, node filtering buffers 100, and result-related subjects buffer 10. These sizes are defined in the OptimizationEvents constructor using ReplaySubject.createWithSize(n). Events beyond the buffer size are discarded from replay. This prevents memory growth in long-running optimizations.

Not a durable store. ReplaySubjects are in-memory buffers. For a durable event history (audit, compliance or analysis after the run), forward the events from your subscription into your logging or telemetry pipeline (OpenTelemetry, ELK, Splunk or a database).

Thread-safe emission. Events are emitted from optimizer threads and consumed in subscriber threads. RxJava handles the thread safety of the subject itself. Keep your subscriber lambda lightweight and non-blocking, so that it does not put backpressure on the optimizer.

Related examples

All in Java-TourOptimizer-Examples on GitHub:

• ChangeOnProgressDefaultFrequencyExample.java: the progress frequency per stage
• RequestOnProgressViaExternalTimeExample.java: progress requests from outside the optimizer
• UncaughtExceptionHandlingExample.java: a custom uncaught exception handler
• UncaughtExceptionHandlingRxJavaWithCustomErrorConsumer.java: an error consumer for the RxJava subscriptions
• Recommended implementation examples: the synchronous, asynchronous and reactive patterns

Related pages

• First optimization: a complete optimization in Java, the base for the code on this page
• AutoFilter: the feature behind the node filtering events
• Optimization properties: the property keys by category, with their defaults
• TourOptimizer REST server: the Spring WebFlux service that runs JOpt.TourOptimizer

Authors

DNA Evolutions GmbH

Added: 2026-03-29 (JOpt core 7.5.3-j17) | Last updated: 2026-10-23