### Example Output of BaseSubscriber Customization

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-backpressure.adoc

This is the expected output when running the `BaseSubscriber` example, showing the request and the subsequent cancellation.

```text
request of 1
Cancelling after having received 1
```

--------------------------------

### Install JDK 21 with SDKMAN!

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Use SDKMAN! to install JDK 21 if it's not already present. This is required for building the project due to Java Multi-Release JAR File support.

```shell
sdk env install

sdk install java $(sdk list java | grep -Eo -m1 '21\b\.[ea|0-9]{1,2}\.[0-9]{1,2}-open')
```

--------------------------------

### Install JDK 9 with SDKMAN!

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Use SDKMAN! to install JDK 9 if it's not already present. This is required for building the project due to Java Multi-Release JAR File support.

```shell
sdk env install

sdk install java $(sdk list java | grep -Eo -m1 '9\b\.[ea|0-9]{1,2}\.[0-9]{1,2}-open')
```

--------------------------------

### Almond.js Optimization Example

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/assets/highlight/README.md

When using Almond.js, you need to use the `r.js` optimizer to specify module names and paths for Highlight.js.

```bash
r.js -o name=hljs paths.hljs=/path/to/highlight out=highlight.js
```

--------------------------------

### Basic Virtual Time Test Setup

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/testing.adoc

Initiates a virtual time test scenario with a Mono that delays for one day. This setup is crucial for testing time-dependent operators without actual delays.

```java
StepVerifier.withVirtualTime(() -> Mono.delay(Duration.ofDays(1)))
//... continue expectations here
```

--------------------------------

### Cold Flux Example

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/reactor-hotCold.adoc

Demonstrates a cold Flux where each subscriber receives all emitted items. This is the default behavior for most Reactor publishers.

```java
Flux<String> source = Flux.fromIterable(Arrays.asList("blue", "green", "orange", "purple"))
                          .map(String::toUpperCase);

source.subscribe(d -> System.out.println("Subscriber 1: "+d));
source.subscribe(d -> System.out.println("Subscriber 2: "+d));
```

--------------------------------

### Mono Example with Operators

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Shows creating a Mono from a callable, using flatMap to combine with another Mono, applying a timeout, and logging the result upon success.

```java
Mono.fromCallable(System::currentTimeMillis)
    .flatMap(time -> Mono.first(serviceA.findRecent(time), serviceB.findRecent(time)))
    .timeout(Duration.ofSeconds(3), errorHandler::fallback)
    .doOnSuccess(r -> serviceM.incrementSuccess())
    .subscribe(System.out::println);
```

--------------------------------

### Build Documentation with PDF Generation

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Build the project documentation and include a PDF version by using the '-PforcePdf' Gradle option. This requires 'asciidoctor-pdf' to be installed.

```shell
./gradlew docs -PforcePdf
```

--------------------------------

### Custom Initialization with jQuery

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/assets/highlight/README.md

This example shows how to manually highlight code blocks using jQuery and the `highlightBlock` function. It iterates over all `pre code` elements.

```javascript
$(document).ready(function() {
  $('pre code').each(function(i, block) {
    hljs.highlightBlock(block);
  });
});
```

--------------------------------

### Java vs. Kotlin Reactor API Examples

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/kotlin.adoc

Compares common Reactor operations in Java with their Kotlin equivalents using extensions.

```java
Mono.just("foo")
```

```kotlin
"foo".toMono()
```

```java
Flux.fromIterable(list)
```

```kotlin
list.toFlux()
```

```java
Mono.error(new RuntimeException())
```

```kotlin
RuntimeException().toMono()
```

```java
Flux.error(new RuntimeException())
```

```kotlin
RuntimeException().toFlux()
```

```java
flux.ofType(Foo.class)
```

```kotlin
flux.ofType<Foo>() or flux.ofType(Foo::class)
```

```java
StepVerifier.create(flux).verifyComplete()
```

```kotlin
flux.test().verifyComplete()
```

--------------------------------

### Predicate-Based Windowing with `Flux.windowWhile()`

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/advanced-three-sorts-batching.adoc

Illustrates predicate-based windowing using `windowWhile`. This example shows how non-matching elements can lead to empty windows, and how completion works.

```java
StepVerifier.create(
	Flux.just(1, 3, 5, 2, 4, 6, 11, 12, 13)
		.windowWhile(i -> i % 2 == 0)
		.concatMap(g -> g.defaultIfEmpty(-1))
	)
		.expectNext(-1, -1, -1) //respectively triggered by odd 1 3 5
		.expectNext(2, 4, 6) // triggered by 11
		.expectNext(12) // triggered by 13
		// however, no empty completion window is emitted (would contain extra matching elements)
		.verifyComplete();
```

--------------------------------

### Flux Example with Operators

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Demonstrates creating a Flux from an iterable, merging with an interval, and applying operators like doOnNext, map, take, onErrorResume, and doAfterTerminate before subscribing.

```java
Flux.fromIterable(getSomeLongList())
    .mergeWith(Flux.interval(100))
    .doOnNext(serviceA::someObserver)
    .map(d -> d * 2)
    .take(3)
    .onErrorResume(errorHandler::fallback)
    .doAfterTerminate(serviceM::incrementTerminate)
    .subscribe(System.out::println);
```

--------------------------------

### Flux.create with Listener and Cancellation

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Example of creating a Flux from an outside context using Flux.create. It includes adding an ActionListener and handling cancellation by removing the listener.

```java
Flux.create(sink -> {
         ActionListener al = e -> {
            sink.next(textField.getText());
         };

         // without cancellation support:
         button.addActionListener(al);

         // with cancellation support:
         sink.onCancel(() -> {
         	button.removeListener(al);
         });
    },
    // Overflow (backpressure) handling, default is BUFFER
    FluxSink.OverflowStrategy.LATEST)
    .timeout(Duration.ofSeconds(3))
    .doOnComplete(() -> System.out.println("completed!"))
    .subscribe(System.out::println);
```

--------------------------------

### Log Errors Imperatively with try-catch

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/error-handling.adoc

This imperative example shows how to catch a RuntimeException, log a message, and re-throw the original error. It serves as a baseline for understanding reactive error handling.

```java
try {
  return callExternalService(k);
}
catch (RuntimeException error) {
  //make a record of the error
  log("uh oh, falling back, service failed for key " + k);
  throw error;
}
```

--------------------------------

### Hot Flux Example using Sinks

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/reactor-hotCold.adoc

Simulates a hot Flux using `Sinks.Many`. New subscribers only receive items emitted after they subscribe. Use `emitNext` or `tryEmitNext` to send data.

```java
Sinks.Many<String> hotSource = Sinks.unsafe().many().multicast().directBestEffort();

Flux<String> hotFlux = hotSource.asFlux().map(String::toUpperCase);

hotFlux.subscribe(d -> System.out.println("Subscriber 1 to Hot Source: "+d));

hotSource.emitNext("blue", FAIL_FAST); // <1>
hotSource.tryEmitNext("green").orThrow(); // <2>

hotFlux.subscribe(d -> System.out.println("Subscriber 2 to Hot Source: "+d));

hotSource.emitNext("orange", FAIL_FAST);
hotSource.emitNext("purple", FAIL_FAST);
hotSource.emitComplete(FAIL_FAST);
```

--------------------------------

### Customizing Initial Request with BaseSubscriber

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-backpressure.adoc

Override `hookOnSubscribe` in `BaseSubscriber` to manually control the initial request. Ensure you call `request` at least once to avoid the Flux getting stuck.

```java
Flux.range(1, 10)
    .doOnRequest(r -> System.out.println("request of " + r))
    .subscribe(new BaseSubscriber<Integer>() {

      @Override
      public void hookOnSubscribe(Subscription subscription) {
        request(1);
      }

      @Override
      public void hookOnNext(Integer integer) {
        System.out.println("Cancelling after having received " + integer);
        cancel();
      }
    });
```

--------------------------------

### HTTP Client Context Propagation Example

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/context.adoc

Shows a library perspective of reading correlation ID from context to augment an HTTP request. It uses deferContextual to access the context and conditionally add a header.

```java
static final String HTTP_CORRELATION_ID = "reactive.http.library.correlationId";

Mono<Tuple2<Integer, String>> doPut(String url, Mono<String> data) {
  Mono<Tuple2<String, Optional<Object>>> dataAndContext = 
      data.zipWith(Mono.deferContextual(c -> // <1>
          Mono.just(c.getOrEmpty(HTTP_CORRELATION_ID))) // <2>
      );

  return dataAndContext.<String>handle((dac, sink) -> {
      if (dac.getT2().isPresent()) { // <3>
        sink.next("PUT <" + dac.getT1() + "> sent to " + url +
            " with header X-Correlation-ID = " + dac.getT2().get());
      }
      else {
        sink.next("PUT <" + dac.getT1() + "> sent to " + url);
      }
        sink.complete();
      })
      .map(msg -> Tuples.of(200, msg));
}
```

--------------------------------

### Using handle to Map and Filter Nulls

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/producing.adoc

An example demonstrating the `handle` operator to map integers to letters using a helper function and filter out any `null` results. This effectively combines mapping and filtering in a single step.

```java
Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
    .handle((i, sink) -> {
        String letter = alphabet(i); // <1>
        if (letter != null) // <2>
            sink.next(letter); // <3>
    });

alphabet.subscribe(System.out::println);
```

--------------------------------

### Reactor Code with Timeout and Fallback

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/reactiveProgramming.adoc

Extends the Reactor example by adding a timeout for the initial operation and a fallback mechanism using a cache service. This showcases Reactor's ability to handle complex scenarios like timeouts and error recovery gracefully.

```java
userService.getFavorites(userId)
           .timeout(Duration.ofMillis(800)) 
           .onErrorResume(cacheService.cachedFavoritesFor(userId)) 
           .flatMap(favoriteService::getDetails) 
           .switchIfEmpty(suggestionService.getSuggestions()) 
           .take(5) 
           .publishOn(UiUtils.uiThreadScheduler()) 
           .subscribe(uiList::show, UiUtils::errorPopup);
```

--------------------------------

### Controller usage of logOnError helper

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

This controller example shows how to integrate the `logOnError` helper function. It combines `logOnNext` for successful events and `logOnError` for error events, ensuring that context is maintained across both scenarios.

```java
@GetMapping("/byPrice")
public Flux<Restaurant> byPrice(@RequestParam Double maxPrice, @RequestHeader(required = false, name = "X-UserId") String userId) {
	String apiId = userId == null ? "" : userId;

	return restaurantService.byPrice(maxPrice))
			   .doOnEach(logOnNext(v -> LOG.info("found restaurant {}", v)))
			   .doOnEach(logOnError(e -> LOG.error("error when searching restaurants", e))) 
			   .contextWrite(Context.of("CONTEXT_KEY", apiId));
}
```

--------------------------------

### Controller usage of logOnNext helper

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

This example demonstrates how to use the `logOnNext` helper function within a reactive web controller. It extracts a user ID from a request header, writes it to the Reactor Context, and then uses `doOnEach` to apply the logging helper.

```java
@GetMapping("/byPrice")
public Flux<Restaurant> byPrice(@RequestParam Double maxPrice, @RequestHeader(required = false, name = "X-UserId") String userId) {
	String apiId = userId == null ? "" : userId; 

	return restaurantService.byPrice(maxPrice))
			   .doOnEach(logOnNext(r -> LOG.debug("found restaurant {} for ${}", 
					r.getName(), r.getPricePerPerson())))
			   .contextWrite(Context.of("CONTEXT_KEY", apiId)); 
}
```

--------------------------------

### Build Project Documentation

Source: https://github.com/reactor/reactor-core/blob/main/README.md

Execute the Gradle wrapper to build the project documentation. The documentation will be generated in the 'docs/build/site/index.html' directory.

```shell
./gradlew docs
```

--------------------------------

### Subscribe with Value Consumer

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-details.adoc

Subscribe to a Flux and provide a lambda to process each emitted value. This example prints each integer to the console.

```java
Flux<Integer> ints = Flux.range(1, 3);
ints.subscribe(i -> System.out.println(i));
```

--------------------------------

### Initialize ReactorDebugAgent

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/debugging.adoc

Explicitly initialize the ReactorDebugAgent to start instrumenting classes. This should ideally be placed early in your application's entry point.

```java
ReactorDebugAgent.init();
```

--------------------------------

### Static Fallback Value with onErrorReturn

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/error-handling.adoc

This example shows the Reactor equivalent of catching an error and returning a static default value using the onErrorReturn operator.

```java
Flux.just(10)
    .map(this::doSomethingDangerous)
    .onErrorReturn("RECOVERED");
```

--------------------------------

### Create Mono and Flux with Factory Methods

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/simple-ways-to-create-a-flux-or-mono-and-subscribe-to-it.adoc

Shows how to create an empty Mono, a Mono with a single value, and a Flux emitting a range of integers using factory methods.

```java
Mono<String> noData = Mono.empty();

Mono<String> data = Mono.just("foo");

Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3);
```

--------------------------------

### Basic Initialization with Highlight.js

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/assets/highlight/README.md

Include these scripts and styles in your HTML to enable automatic syntax highlighting on page load. It works with `<pre><code>` tags and auto-detects languages.

```html
<link rel="stylesheet" href="/path/to/styles/default.css">
<script src="/path/to/highlight.pack.js"></script>
<script>hljs.initHighlightingOnLoad();</script>
```

--------------------------------

### Enable Global Operator Debugging

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/debugging.adoc

Activate assembly-time instrumentation for debugging by calling `Hooks.onOperatorDebug()` at the start of your application. This should be done before any `Flux` or `Mono` instances are created.

```java
Hooks.onOperatorDebug();
```

--------------------------------

### Attach SampleSubscriber to a Flux

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-details.adoc

Demonstrates how to attach a custom `SampleSubscriber` to a `Flux` to process its elements.

```java
SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
Flux<Integer> ints = Flux.range(1, 4);
ints.subscribe(ss);
```

--------------------------------

### Manual Connection with ConnectableFlux

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/advanced-broadcast-multiple-subscribers-connectableflux.adoc

Use publish() to create a ConnectableFlux and manually call connect() after setting up multiple subscribers. This triggers the subscription to the upstream source.

```java
Flux<Integer> source = Flux.range(1, 3)
                           .doOnSubscribe(s -> System.out.println("subscribed to source"));

ConnectableFlux<Integer> co = source.publish();

co.subscribe(System.out::println, e -> {}, () -> {});
co.subscribe(System.out::println, e -> {}, () -> {});

System.out.println("done subscribing");
Thread.sleep(500);
System.out.println("will now connect");

co.connect();
```

--------------------------------

### Using TupleUtils with Lambda Expressions in Java

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/apdx-reactorExtra.adoc

Bridge between Java 8 functional interfaces and Tuple operations using TupleUtils. This example maps a Tuple to a Customer object.

```java
.map(tuple -> {
  String firstName = tuple.getT1();
  String lastName = tuple.getT2();
  String address = tuple.getT3();

  return new Customer(firstName, lastName, address);
});
```

```java
.map(TupleUtils.function(Customer::new));
```

--------------------------------

### Subscribing to a Flux with Immediate Scheduler

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/debugging.adoc

Demonstrates how to subscribe to a Flux, specifying the immediate scheduler for execution. This snippet is used to illustrate a scenario where the Flux declaration is separated from its subscription point, complicating debugging.

```java
toDebug
    .subscribeOn(Schedulers.immediate())
    .subscribe(System.out::println, Throwable::printStackTrace);
```

--------------------------------

### Create a Replay Sink

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/processors.adoc

Instantiates a Sinks.Many that replays all emitted data to new subscribers.

```java
Sinks.Many<Integer> replaySink = Sinks.many().replay().all();
```

--------------------------------

### Callback Hell Example in Java

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/reactiveProgramming.adoc

Illustrates a complex asynchronous operation using nested callbacks, leading to difficult-to-read and maintain code. This pattern is often referred to as 'Callback Hell'.

```java
userService.getFavorites(userId, new Callback<List<String>>() { 
  public void onSuccess(List<String> list) { 
    if (list.isEmpty()) { 
      suggestionService.getSuggestions(new Callback<List<Favorite>>() { 
        public void onSuccess(List<Favorite> list) { 
          UiUtils.submitOnUiThread(() -> { 
            list.stream() 
                .limit(5) 
                .forEach(uiList::show); 
            }); 
        } 

        public void onError(Throwable error) { 
          UiUtils.errorPopup(error); 
        } 
      }); 
    } else { 
      list.stream() 
          .limit(5) 
          .forEach(favId -> favoriteService.getDetails(favId, 
            new Callback<Favorite>() { 
              public void onSuccess(Favorite details) { 
                UiUtils.submitOnUiThread(() -> uiList.show(details)); 
              } 

              public void onError(Throwable error) { 
                UiUtils.errorPopup(error); 
              } 
            } 
          )); 
    } 
  } 

  public void onError(Throwable error) { 
    UiUtils.errorPopup(error); 
  } 
});
```

--------------------------------

### Switch Execution Contexts with publishOn

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

Demonstrates how to use multiple publishOn operators to switch execution contexts for different stages of a Flux processing chain. The placement of publishOn is significant.

```java
Sinks.Many<Integer> dataSinks = Sinks.many().unicast().onBackpressureBuffer();
Flux<Integer> source = dataSinks.asFlux();
source.publishOn(scheduler1)
	  .map(i -> transform(i))
	  .publishOn(scheduler2)
	  .doOnNext(i -> processNext(i))
	  .subscribe();
```

--------------------------------

### Run Mono in a New Thread

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/schedulers.adoc

Demonstrates how to subscribe to a Mono on a new thread. The Mono is assembled on the main thread but its operations and subscription callbacks execute on the newly created thread.

```java
public static void main(String[] args) throws InterruptedException {
  final Mono<String> mono = Mono.just("hello "); //<1>

  Thread t = new Thread(() -> mono
      .map(msg -> msg + "thread ")
      .subscribe(v -> //<2>
          System.out.println(v + Thread.currentThread().getName()) //<3>
      )
  );
  t.start();
  t.join();

}
```

--------------------------------

### Custom Retry Logic with retryWhen

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/error-handling.adoc

Implement advanced retry strategies using retryWhen, which allows customization of retry conditions based on a companion Flux. This example emulates retry(3).

```java
Flux<String> flux = Flux
    .<String>error(new IllegalArgumentException())
    .doOnError(System.out::println)
    .retryWhen(Retry.from(companion -> 
        companion.take(3)));
```

--------------------------------

### Test Command Empty Path with PublisherProbe

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/testing.adoc

Uses PublisherProbe to verify that the fallback path (doWhenEmpty) is executed when the command source is empty. Asserts subscription, request, and non-cancellation.

```java
@Test
public void testCommandEmptyPathIsUsed() {
    PublisherProbe<Void> probe = PublisherProbe.empty(); // <1>

    StepVerifier.create(processOrFallback(Mono.empty(), probe.mono())) // <2>
                .verifyComplete();

    probe.assertWasSubscribed(); //<3>
    probe.assertWasRequested(); //<4>
    probe.assertWasNotCancelled(); //<5>
}
```

--------------------------------

### Flux.create with Callbacks

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/producing.adoc

Demonstrates setting up callbacks for cancel and dispose signals within a `Flux.create` producer. `onCancel` is invoked for cancel signals, while `onDispose` handles complete, error, or cancel signals.

```java
Flux<String> bridge = Flux.create(sink -> {
    sink.onRequest(n -> channel.poll(n))
        .onCancel(() -> channel.cancel()) // <1>
        .onDispose(() -> channel.close());  // <2>
    });
```

--------------------------------

### Overlapping Windows with `Flux.window()`

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/advanced-three-sorts-batching.adoc

Demonstrates creating overlapping windows using `window(maxSize, skip)`. When `maxSize` is greater than `skip`, new windows open before previous ones close. Empty windows are shown as -1.

```java
StepVerifier.create(
	Flux.range(1, 10)
		.window(5, 3) //overlapping windows
		.concatMap(g -> g.defaultIfEmpty(-1)) //show empty windows as -1
	)
		.expectNext(1, 2, 3, 4, 5)
		.expectNext(4, 5, 6, 7, 8)
		.expectNext(7, 8, 9, 10)
		.expectNext(10)
		.verifyComplete();
```

--------------------------------

### Use subscribeOn to switch subscription context

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/schedulers.adoc

Demonstrates switching the execution context for the entire chain, including subscription, using subscribeOn. All operations, including the first map, run on threads from the provided parallel scheduler.

```java
Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); //<1>

final Flux<String> flux = Flux
    .range(1, 2)
    .map(i -> 10 + i)  //<2>
    .subscribeOn(s)  //<3>
    .map(i -> "value " + i);  //<4>

new Thread(() -> flux.subscribe(System.out::println));  //<5>
```

--------------------------------

### Test zip with Optional to Ensure All Subscribed

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

Ensures subscription to all publishers when using zip with empty-completed publishers by wrapping each publisher with `singleOptional`. This preserves the semantics of subscribing to all sources.

```java

@Test
public void testZipOptionalAllSubscribed() {
	AtomicInteger cnt = new AtomicInteger();
	Mono<Integer> mono1 = Mono.create(sink -> {
		cnt.incrementAndGet();
		sink.success();
	});
	Mono<Integer> mono2 = Mono.create(sink -> {
		cnt.incrementAndGet();
		sink.success();
	});
	Mono<Integer> mono3 = Mono.create(sink -> {
		cnt.incrementAndGet();
		sink.success();
	});
	Mono<Optional<Integer>> zippedMono =
			Mono.zip(
					mono1.singleOptional(),
					Mono.zip(mono2.singleOptional(), mono3.singleOptional(), (v1, v2) -> v1),
					(v1, v2) -> v1);
	zippedMono.subscribe();
	assertEquals(3, cnt.get());
}
```

--------------------------------

### Typical Reactor Stack Trace Example

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/debugging.adoc

Illustrates a common stack trace encountered in Reactor, specifically an IndexOutOfBoundsException from a MonoSingle operator indicating a violation of the 'exactly one item' contract. This helps in understanding how to read and interpret asynchronous error messages.

```java
java.lang.IndexOutOfBoundsException: Source emitted more than one item
    at reactor.core.publisher.MonoSingle$SingleSubscriber.onNext(MonoSingle.java:129)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.tryEmitScalar(FluxFlatMap.java:445)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onNext(FluxFlatMap.java:379)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:121)
    at reactor.core.publisher.FluxRange$RangeSubscription.slowPath(FluxRange.java:154)
    at reactor.core.publisher.FluxRange$RangeSubscription.request(FluxRange.java:109)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
    at reactor.core.publisher.FluxFlatMap$FlatMapMain.onSubscribe(FluxFlatMap.java:332)
    at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onSubscribe(FluxMapFuseable.java:90)
    at reactor.core.publisher.FluxRange.subscribe(FluxRange.java:68)
    at reactor.core.publisher.FluxMapFuseable.subscribe(FluxMapFuseable.java:63)
    at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:97)
    at reactor.core.publisher.MonoSingle.subscribe(MonoSingle.java:58)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3096)
    at reactor.core.publisher.Mono.subscribeWith(Mono.java:3204)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3090)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3057)
    at reactor.core.publisher.Mono.subscribe(Mono.java:3029)
    at reactor.guide.GuideTests.debuggingCommonStacktrace(GuideTests.java:995)
```

--------------------------------

### SampleSubscriber Output

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-details.adoc

Illustrates the expected output when using the `SampleSubscriber` with a `Flux` of integers from 1 to 4.

```text
Subscribed
1
2
3
4
```

--------------------------------

### Group Strings by First Character with groupBy and Prefetch

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/advanced-three-sorts-batching.adoc

Groups strings by their first character, specifying a prefetch value for `groupBy`. Includes logging within the processing chain to observe element flow. This example highlights potential deadlocks if groups are not consumed efficiently.

```java
public static Flux<String> createGroupedFlux() {
        List<String> data = List.of("alpha", "air", "aim", "beta", "cat", "ball", "apple", "bat", "dog", "ace");
        return Flux.fromIterable(data)
                .groupBy(d -> d.charAt(0), 5)
                .concatMap(g -> g.map(String::valueOf)
                        .startWith(String.valueOf(g.key()))
                        .map(o -> {
                            System.out.println(o);
                            return o;
                        })
                );
    }

    @Test
    public void testGroupBy() {
        StepVerifier.create(createGroupedFlux())
                .expectNext("a", "alpha", "air", "aim", "apple", "ace")
                .expectNext("b", "beta", "ball", "bat")
                .expectNext("c", "cat", "d", "dog")
                .verifyComplete();
    }
```

--------------------------------

### Emulate retry(3) with retryWhen

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

Demonstrates how to use the `retryWhen` operator to replicate the behavior of a simple `retry(3)` call. This is useful for understanding the more complex `retryWhen` operator.

```java
include::snippetRetryWhenRetry.adoc[]
```

--------------------------------

### Wrap Scheduler with Metrics

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/metrics.adoc

Wrap an existing Scheduler with Micrometer metrics to measure task execution. Provide the Scheduler, MeterRegistry, a meter name prefix, and optional tags.

```java
Scheduler originalScheduler = Schedulers.newParallel("test", 4);

Scheduler schedulerWithMetrics = Micrometer.timedScheduler(
	originalScheduler, // <1>
	applicationDefinedMeterRegistry, // <2>
	"testingMetrics", // <3>
	Tags.of(Tag.of("additionalTag", "yes")) // <4>
);

```

--------------------------------

### Create Flux from Strings and Iterable

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/simple-ways-to-create-a-flux-or-mono-and-subscribe-to-it.adoc

Demonstrates creating a Flux from a list of strings using `Flux.just` and `Flux.fromIterable`.

```java
Flux<String> seq1 = Flux.just("foo", "bar", "foobar");

List<String> iterable = Arrays.asList("foo", "bar", "foobar");
Flux<String> seq2 = Flux.fromIterable(iterable);
```

--------------------------------

### Use publishOn to switch execution context

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/schedulers.adoc

Demonstrates switching the execution context for subsequent operators using publishOn. The first map runs on the subscription thread, while the second map and subsequent operations run on threads from the provided parallel scheduler.

```java
Scheduler s = Schedulers.newParallel("parallel-scheduler", 4); //<1>

final Flux<String> flux = Flux
    .range(1, 2)
    .map(i -> 10 + i)  //<2>
    .publishOn(s)  //<3>
    .map(i -> "value " + i);  //<4>

new Thread(() -> flux.subscribe(System.out::println));  //<5>
```

--------------------------------

### Highlighting with Web Workers (Main Script)

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/assets/highlight/README.md

This script demonstrates how to use a Web Worker for syntax highlighting to prevent the browser from freezing with large code blocks. It sends the code content to the worker and receives the highlighted HTML.

```javascript
addEventListener('load', function() {
  var code = document.querySelector('#code');
  var worker = new Worker('worker.js');
  worker.onmessage = function(event) { code.innerHTML = event.data; }
  worker.postMessage(code.textContent);
})
```

--------------------------------

### Using `single()` vs. `take(1)` in Reactor

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/debugging.adoc

Illustrates a common debugging scenario where `single()` is too restrictive and `take(1)` is a more appropriate alternative for Flux operations.

```java
private Mono<String> scatterAndGather(Flux<String> urls) {
    return urls.flatMap(url -> doRequest(url))
           .single(); 
}
```

--------------------------------

### Minimalistic BaseSubscriber Implementation

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/subscribe-details.adoc

Provides a basic implementation of `BaseSubscriber` for custom subscription logic, including handling subscription, next events, and requesting elements.

```java
package io.projectreactor.samples;

import org.reactivestreams.Subscription;

import reactor.core.publisher.BaseSubscriber;

public class SampleSubscriber<T> extends BaseSubscriber<T> {

	@Override
	public void hookOnSubscribe(Subscription subscription) {
		System.out.println("Subscribed");
		request(1);
	}

	@Override
	public void hookOnNext(T value) {
		System.out.println(value);
		request(1);
	}
}
```

--------------------------------

### Imperative Resource Management with try-with-resource

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/error-handling.adoc

Shows the Java 7 try-with-resource construct for automatically managing resources that implement AutoCloseable.

```java
try (SomeAutoCloseable disposableInstance = new SomeAutoCloseable()) {
  return disposableInstance.toString();
}
```

--------------------------------

### Testing Context Propagation with StepVerifier

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/testing.adoc

Use StepVerifierOptions to provide an initial Context and assert its propagation. Call then() to return to sequence expectations.

```java
StepVerifier.create(Mono.just(1).map(i -> i + 10),
				StepVerifierOptions.create().withInitialContext(Context.of("thing1", "thing2"))) // <1>
		            .expectAccessibleContext()//<2>
		            .contains("thing1", "thing2") // <3>
		            .then() // <4>
		            .expectNext(11)
		            .verifyComplete(); // <5>
```

--------------------------------

### Reading and Writing to Reactor Context

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/context.adoc

Demonstrates how to read a value from the context and write a new value into it within a reactive sequence. Use `contextWrite` to provide context data and `deferContextual` to access it.

```java
String key = "message";
Mono<String> r = Mono.just("Hello")
    .flatMap(s -> Mono.deferContextual(ctx ->
         Mono.just(s + " " + ctx.get(key))))
    .contextWrite(ctx -> ctx.put(key, "World"));

StepVerifier.create(r)
            .expectNext("Hello World")
            .verifyComplete();
```

--------------------------------

### Test zip with Empty Completion (All Subscribed)

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/faq.adoc

Demonstrates a scenario where Mono.zip subscribes to all provided publishers when they complete empty. This test case verifies the behavior when all zipped Monos complete without emitting items.

```java
    @Test
    public void testZipEmptyCompletionAllSubscribed() {
        AtomicInteger cnt = new AtomicInteger();
        Mono<Integer> mono1 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Integer> mono2 = Mono.create(sink -> {
            cnt.incrementAndGet();
            sink.success();
        });
        Mono<Integer> zippedMono = Mono.zip(mono1, mono2, (v1, v2) -> v1);
        zippedMono.subscribe();
        assertEquals(2, cnt.get());
    }
```

--------------------------------

### Imperative Resource Cleanup with finally

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/coreFeatures/error-handling.adoc

Demonstrates the imperative approach to resource cleanup using a finally block, ensuring that operations like stopping timers are always executed.

```java
Stats stats = new Stats();
stats.startTimer();
try {
  doSomethingDangerous();
}
finally {
  stats.stopTimerAndRecordTiming();
}
```

--------------------------------

### Context Propagation with flatMap

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/advancedFeatures/context.adoc

Demonstrates how contextWrite writes are scoped and how flatMap affects context visibility. The context written within a flatMap is not visible outside of it.

```java
String key = "message";
Mono<String> r = Mono
    .deferContextual(ctx -> Mono.just("Hello " + ctx.get(key))) //<3>
    .contextWrite(ctx -> ctx.put(key, "Reactor")) //<2>
    .flatMap( s -> Mono.deferContextual(ctx ->
        Mono.just(s + " " + ctx.get(key)))) //<4>
    .contextWrite(ctx -> ctx.put(key, "World")); //<1>

StepVerifier.create(r)
            .expectNext("Hello Reactor World") //<5>
            .verifyComplete();
```

--------------------------------

### Test Empty Path Execution

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/testing.adoc

Tests the scenario where the source publisher is empty, triggering the fallback.

```java
@Test
public void testEmptyPathIsUsed() {
    StepVerifier.create(processOrFallback(Mono.empty(), Mono.just("EMPTY_PHRASE")))
                .expectNext("EMPTY_PHRASE")
                .verifyComplete();
}
```

--------------------------------

### Observe Reactive Chains with Micrometer Observation

Source: https://github.com/reactor/reactor-core/blob/main/docs/modules/ROOT/pages/metrics.adoc

Instrument a reactive chain using `Micrometer.observation` via the `tap` operator. This provides a lifecycle-based observation, potentially translating to timers, spans, or logs.

```java
listenToEvents()
    .name("events") // <1>
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .tap(Micrometer.observation( // <2>
		applicationDefinedRegistry)) // <3>
    .retry()
    .subscribe();
```