anchor.diff:  
   import java.math.BigDecimal;
   import java.math.RoundingMode;
  +import java.time.Duration;
   import java.time.Instant;
   import java.util.Arrays;
   import java.util.Collections;
   import java.util.List;
   import java.util.Random;
  -import java.util.concurrent.TimeUnit;
  -import java.util.stream.Collectors;
   
   import javax.enterprise.context.ApplicationScoped;
   
  +import io.smallrye.mutiny.Multi;
  +import io.smallrye.reactive.messaging.kafka.Record;
   import org.eclipse.microprofile.reactive.messaging.Outgoing;
   import org.jboss.logging.Logger;
  -
  -import io.reactivex.Flowable;
  -import io.smallrye.reactive.messaging.kafka.KafkaRecord;
   
   /**
    * A bean producing random temperature data every second.
  
positive.code:     }
  }
  ----
  <1> Instruct Reactive Messaging to dispatch the items from the returned `Flowable` to `temperature-values`.
  <2> The method returns a RX Java 2 _stream_ (`Flowable`) emitting a random temperature value every 0.5 seconds.
  <3> Instruct Reactive Messaging to dispatch the items from the returned `Flowable` (static list of weather stations) to `weather-stations`.
  
  The two methods each return a _reactive stream_ whose items are sent to the streams named `temperature-values` and `weather-stations`, respectively.
  
  
negative.code:                             .doubleValue();
  
                      LOG.infov("station: {0}, temperature: {1}", station.name, temperature);
                      return Record.of(station.id, Instant.now() + ";" + temperature);
                  });
      }
  
      @Outgoing("weather-stations")                                          // <3>
      public Multi<Record<Integer, String>> weatherStations() {
          return Multi.createFrom().items(stations.stream()
              .map(s -> Record.of(
                      s.id,
                      "{ \"id\" : " + s.id +
                      ", \"name\" : \"" + s.name + "\" }"))
          );
      }
  
      private static class WeatherStation {
  
  