Add examples for reactive streams

examples/pom.xml

@@ -141,6 +141,23 @@ limitations under the License.
             <scope>runtime</scope>
         </dependency>
 
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.reactivex.rxjava3</groupId>
+            <artifactId>rxjava</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.smallrye.reactive</groupId>
+            <artifactId>mutiny</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>net.automatalib</groupId>
             <artifactId>automata-api</artifactId>

examples/src/main/java/de/learnlib/example/reactive/MutinyExample.java

@@ -0,0 +1,130 @@
+/* Copyright (C) 2013-2026 TU Dortmund University
+ * This file is part of LearnLib <https://learnlib.de>.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package de.learnlib.example.reactive;
+
+import java.util.Objects;
+import java.util.Random;
+import java.util.concurrent.Executors;
+
+import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy;
+import de.learnlib.driver.simulator.MealySimulatorSUL;
+import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder;
+import de.learnlib.oracle.equivalence.RandomWMethodEQOracle;
+import de.learnlib.oracle.parallelism.ParallelOracleBuilders;
+import de.learnlib.query.DefaultQuery;
+import io.smallrye.mutiny.Multi;
+import net.automatalib.alphabet.impl.Alphabets;
+import net.automatalib.automaton.transducer.MealyMachine;
+import net.automatalib.util.automaton.Automata;
+import net.automatalib.util.automaton.random.RandomAutomata;
+import net.automatalib.word.Word;
+
+/**
+ * An example of constructing a learn-loop using reactive streams (from SmallRye) to compute counterexamples.
+ */
+// allow println and vars in examples, ExecutorService does not implement AutoClosable until Java 19+
+@SuppressWarnings("PMD")
+public final class MutinyExample {
+
+    private static final int SEED = 42;
+    private static final int SIZE = 10;
+    private static final int NUM_INPUTS = 4;
+    private static final int RND_LENGTH = 4;
+    private static final int LIMIT = 1000;
+
+    private MutinyExample() {
+        // prevent instantiation
+    }
+
+    public static void main(String[] args) {
+        // setup symbols
+        var inputs = Alphabets.integers(0, NUM_INPUTS);
+        var outputs = Alphabets.characters('a', 'd');
+
+        // setup membership oracle
+        var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs);
+        var sul = new MealySimulatorSUL<>(mealy);
+        // IMPORTANT: make sure to use parallel-aware oracle (with thread local instances)
+        // because it will be called from different threads in the reactive environment
+        var mqo = ParallelOracleBuilders.newDynamicParallelOracle(sul).create();
+
+        // setup equivalence oracles
+        var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH);
+        var eqo2 =
+                new KWayStateCoverEQOracleBuilder<MealyMachine<?, Integer, ?, Character>, Integer, Word<Character>>().withOracle(
+                        mqo).withRandom(new Random(SEED)).create();
+
+        // setup learner
+        var learner = new TTTLearnerMealy<>(inputs, mqo);
+
+        // setup thread pools
+        var pool = Executors.newFixedThreadPool(1);
+        var pool2 = Executors.newFixedThreadPool(1);
+        var pool3 = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
+
+        // learning loop
+        learner.startLearning();
+        var hyp = learner.getHypothesisModel();
+
+        while (true) {
+            // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads
+            final var finalHyp = hyp;
+            var m1 = Multi.createFrom()
+                          // create a (cold) publisher from first quivalence oracle
+                          .items(() -> eqo.generateTestWords(finalHyp, inputs))
+                          // sample on own thread
+                          .runSubscriptionOn(pool)
+                          // limit to 1000 elements
+                          .select().first(LIMIT);
+            var m2 = Multi.createFrom()
+                          // create a (cold) publisher from second quivalence oracle
+                          .items(() -> eqo2.generateTestWords(finalHyp, inputs))
+                          // sample on own thread
+                          .runSubscriptionOn(pool2);
+
+            var ce = Multi.createBy()
+                          // merge elements from both oracles in interleaving fashion
+                          .merging().streams(m1, m2)
+                          // run processing in parallel
+                          .runSubscriptionOn(pool3)
+                          // filter for counterexamples
+                          .filter(w -> !Objects.equals(mqo.answerQuery(w), finalHyp.computeOutput(w)))
+                          // toUni implicitly fetches the first element
+                          .toUni()
+                          // use a blocking call to extract the final counterexample
+                          .await().indefinitely();
+
+            if (ce != null) {
+                learner.refineHypothesis(new DefaultQuery<>(ce, mqo.answerQuery(ce)));
+                hyp = learner.getHypothesisModel();
+            } else {
+                break;
+            }
+        }
+
+        // cleanup
+        mqo.shutdown();
+        pool.shutdown();
+        pool2.shutdown();
+        pool3.shutdown();
+
+        // process results
+        hyp = learner.getHypothesisModel();
+
+        System.out.println("Final hypothesis size " + hyp.size());
+        System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs));
+    }
+}

examples/src/main/java/de/learnlib/example/reactive/RXJavaExample.java

@@ -0,0 +1,129 @@
+/* Copyright (C) 2013-2026 TU Dortmund University
+ * This file is part of LearnLib <https://learnlib.de>.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package de.learnlib.example.reactive;
+
+import java.util.Objects;
+import java.util.Random;
+
+import de.learnlib.algorithm.ttt.mealy.TTTLearnerMealy;
+import de.learnlib.driver.simulator.MealySimulatorSUL;
+import de.learnlib.oracle.equivalence.KWayStateCoverEQOracleBuilder;
+import de.learnlib.oracle.equivalence.RandomWMethodEQOracle;
+import de.learnlib.oracle.parallelism.ParallelOracleBuilders;
+import de.learnlib.query.DefaultQuery;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import net.automatalib.alphabet.impl.Alphabets;
+import net.automatalib.automaton.transducer.MealyMachine;
+import net.automatalib.util.automaton.Automata;
+import net.automatalib.util.automaton.random.RandomAutomata;
+import net.automatalib.word.Word;
+
+/**
+ * An example of constructing a learn-loop using reactive streams (from RXJava) to compute counterexamples.
+ */
+@SuppressWarnings({"PMD.SystemPrintln", "PMD.UseExplicitTypes"}) // allow println and vars in examples
+public final class RXJavaExample {
+
+    private static final int SEED = 42;
+    private static final int SIZE = 10;
+    private static final int BATCH_SIZE = 10;
+    private static final int NUM_INPUTS = 4;
+    private static final int RND_LENGTH = 4;
+    private static final int LIMIT = 1000;
+
+    private RXJavaExample() {
+        // prevent instantiation
+    }
+
+    public static void main(String[] args) {
+        // setup symbols
+        var inputs = Alphabets.integers(0, NUM_INPUTS);
+        var outputs = Alphabets.characters('a', 'd');
+
+        // setup membership oracle
+        var mealy = RandomAutomata.randomMealy(new Random(SEED), SIZE, inputs, outputs);
+        var sul = new MealySimulatorSUL<>(mealy);
+        // note that we can still use a parallel oracle to answer query batches in parallel
+        var mqo = ParallelOracleBuilders.newStaticParallelOracle(sul)
+                                        .withNumInstances(BATCH_SIZE)
+                                        .withMinBatchSize(1)
+                                        .create();
+
+        // setup equivalence oracles
+        var eqo = new RandomWMethodEQOracle<>(mqo, SIZE / 2, RND_LENGTH);
+        var eqo2 =
+                new KWayStateCoverEQOracleBuilder<MealyMachine<?, Integer, ?, Character>, Integer, Word<Character>>().withOracle(
+                        mqo).withRandom(new Random(SEED)).create();
+
+        // setup learner
+        var learner = new TTTLearnerMealy<>(inputs, mqo);
+
+        // learning loop
+        learner.startLearning();
+        var hyp = learner.getHypothesisModel();
+
+        while (true) {
+            // since we only access the hypothesis in read-only fashion it is fine to share the reference across threads
+            final var finalHyp = hyp;
+            var ce = Flowable // create a (cold) publisher from first quivalence oracle
+                              .defer(() -> Flowable.fromStream(eqo.generateTestWords(finalHyp, inputs)))
+                              // sample on own thread
+                              .subscribeOn(Schedulers.computation())
+                              // limit to 1000 elements
+                              .take(LIMIT)
+                              // merge with elements from second equivalence oracle, also sampled in its own thread
+                              .mergeWith(Flowable.defer(() -> Flowable.fromStream(eqo2.generateTestWords(finalHyp,
+                                                                                                         inputs)))
+                                                 .subscribeOn(Schedulers.computation()))
+                              // map to queries ...
+                              .map(DefaultQuery<Integer, Word<Character>>::new)
+                              // ... create batches ...
+                              .buffer(BATCH_SIZE)
+                              // ... and process in bulk
+                              // NOTE: this happens synchronously to allow the oracle/SUL to gracefully shutdown
+                              // There exist ways to create schedulers that do not interrupt running threads
+                              // (Schedulers#from) but I still experienced the occasional race condition ...
+                              // However, the oracle can still answer the batch in parallel itself
+                              .doOnNext(mqo::processQueries)
+                              // flat to individual queries
+                              .flatMapIterable(l -> l)
+                              // filter for counterexamples
+                              .filter(q -> !Objects.equals(q.getOutput(),
+                                                           finalHyp.computeSuffixOutput(q.getPrefix(), q.getSuffix())))
+                              // the first counterexample suffices for refinement
+                              .firstElement()
+                              // use a blocking operation to ensure that all pipelines are cleared for the next iteration
+                              .blockingGet();
+
+            if (ce != null) {
+                learner.refineHypothesis(ce);
+                hyp = learner.getHypothesisModel();
+            } else {
+                break;
+            }
+        }
+
+        // cleanup
+        mqo.shutdown();
+
+        // process results
+        hyp = learner.getHypothesisModel();
+
+        System.out.println("Final hypothesis size " + hyp.size());
+        System.out.println("Is equivalent? " + Automata.testEquivalence(mealy, hyp, inputs));
+    }
+}