final var scheduler = Executors.newScheduledThreadPool(1);
final var nullifier = new FitnessNullifier<ProgramGene<Double>, Double>();
final var sampling = new SampleBuffer<Double>(100);
scheduler.scheduleWithFixedDelay(
    () -> {
        // Adding a new sample point every second to the ring buffer.
        sampling.add(nextSamplePoint());
        // Force re-evaluation of populations fitness values.
        nullifier.nullifyFitness();
    },
    1, 1, TimeUnit.SECONDS
);

final Codec<Tree<Op<Double>, ?>, ProgramGene<Double>> codec =
    Regression.codecOf(OPS, TMS, 5, t -> t.gene().size() < 30);

final Regression<Double> regression = Regression.of(
    codec,
    Error.of(LossFunction::mse),
    sampling
);

final Engine<ProgramGene<Double>, Double> engine = Engine
    .builder(regression)
    .interceptor(nullifier)
    .build();

engine.stream()
    .flatMap(Streams.toIntervalMax(Duration.ofSeconds(30)))
    .map(program -> program.bestPhenotype()
        .genotype().gene()
        .toParenthesesString())
    // Printing the best program found so far every 30 seconds.
    .forEach(System.out::println);

final ISeq<Integer> values = IntStream.range(0, streamSize).boxed()
    .flatMap(Streams.toIntervalMax(sliceSize))
    .collect(ISeq.toISeq());

final var publisher = new StreamPublisher<EvolutionResult<IntegerGene, Integer>>();
try (publisher) {
    final var stream= engine.stream();
    publisher.subscribe(new Subscriber<>() { ... });
    publisher.attach(stream);
    ...
}