@@ -780,7 +784,7 @@
* available backpressure modes
* @param emitter Consume the {@link FluxSink} provided per-subscriber by Reactor to generate signals.
* @return a {@link Flux}
- * @see #create(Consumer, OverflowStrategy)
+ * @see #create(Consumer, reactor.core.publisher.FluxSink.OverflowStrategy)
*/
public static Flux push(Consumer> emitter, OverflowStrategy backpressure) {
return onAssembly(new FluxCreate<>(emitter, backpressure, FluxCreate.CreateMode.PUSH_ONLY));
@@ -1060,7 +1064,7 @@
*/
public static Flux from(Publisher source) {
//duplicated in wrap, but necessary to detect early and thus avoid applying assembly
- if (source instanceof Flux) {
+ if (source instanceof Flux && !ContextPropagationSupport.shouldWrapPublisher(source)) {
@SuppressWarnings("unchecked")
Flux casted = (Flux) source;
return casted;
@@ -2130,6 +2134,64 @@
}
/**
+ * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+ * while streaming the values from a Publisher derived from the same resource and makes sure
+ * the resource is released if the sequence terminates or the Subscriber cancels.
+ *
+ * Eager {@link AutoCloseable} resource cleanup happens just before the source termination and exceptions raised
+ * by the cleanup Consumer may override the terminal event.
+ *
+ *
+ *
+ * For an asynchronous version of the cleanup, with distinct path for onComplete, onError
+ * and cancel terminations, see {@link #usingWhen(Publisher, Function, Function, BiFunction, Function)}.
+ *
+ * @param resourceSupplier a {@link Callable} that is called on subscribe to generate the resource
+ * @param sourceSupplier a factory to derive a {@link Publisher} from the supplied resource
+ * @param emitted type
+ * @param resource type
+ *
+ * @return a new {@link Flux} built around a disposable resource
+ * @see #usingWhen(Publisher, Function, Function, BiFunction, Function)
+ * @see #usingWhen(Publisher, Function, Function)
+ */
+ public static Flux using(Callable resourceSupplier,
+ Function> sourceSupplier) {
+ return using(resourceSupplier, sourceSupplier, true);
+ }
+
+ /**
+ * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+ * while streaming the values from a Publisher derived from the same resource and makes sure
+ * the resource is released if the sequence terminates or the Subscriber cancels.
+ *
+ *
+ *
Eager {@link AutoCloseable} resource cleanup happens just before the source termination and exceptions raised
+ * by the cleanup Consumer may override the terminal event.
+ *
Non-eager cleanup will drop any exception.
+ *
+ *
+ *
+ *
+ * For an asynchronous version of the cleanup, with distinct path for onComplete, onError
+ * and cancel terminations, see {@link #usingWhen(Publisher, Function, Function, BiFunction, Function)}.
+ *
+ * @param resourceSupplier a {@link Callable} that is called on subscribe to generate the resource
+ * @param sourceSupplier a factory to derive a {@link Publisher} from the supplied resource
+ * @param eager true to clean before terminating downstream subscribers
+ * @param emitted type
+ * @param resource type
+ *
+ * @return a new {@link Flux} built around a disposable resource
+ * @see #usingWhen(Publisher, Function, Function, BiFunction, Function)
+ * @see #usingWhen(Publisher, Function, Function)
+ */
+ public static Flux using(Callable resourceSupplier,
+ Function> sourceSupplier, boolean eager) {
+ return using(resourceSupplier, sourceSupplier, Exceptions.AUTO_CLOSE, eager);
+ }
+
+ /**
* Uses a resource, generated by a {@link Publisher} for each individual {@link Subscriber},
* while streaming the values from a {@link Publisher} derived from the same resource.
* Whenever the resulting sequence terminates, a provided {@link Function} generates
@@ -2709,7 +2771,8 @@
* signals its first value, completes or a timeout expires. Returns that value,
* or null if the Flux completes empty. In case the Flux errors, the original
* exception is thrown (wrapped in a {@link RuntimeException} if it was a checked
- * exception). If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * exception). If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
*
* Note that each blockFirst() will trigger a new subscription: in other words,
* the result might miss signal from hot publishers.
@@ -2718,6 +2781,7 @@
*
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
* @return the first value or null
*/
@Nullable
@@ -2759,7 +2823,8 @@
* signals its last value, completes or a timeout expires. Returns that value,
* or null if the Flux completes empty. In case the Flux errors, the original
* exception is thrown (wrapped in a {@link RuntimeException} if it was a checked
- * exception). If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * exception). If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
*
* Note that each blockLast() will trigger a new subscription: in other words,
* the result might miss signal from hot publishers.
@@ -2768,6 +2833,7 @@
*
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
* @return the last value or null
*/
@Nullable
@@ -5054,7 +5120,7 @@
}
/**
- * Map this {@link Flux} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Flux} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time
* between each signal as measured by the {@link Schedulers#parallel() parallel} scheduler.
* First duration is measured between the subscription and the first element.
@@ -5071,7 +5137,7 @@
}
/**
- * Map this {@link Flux} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Flux} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time
* between each signal as measured by the provided {@link Scheduler}.
* First duration is measured between the subscription and the first element.
@@ -6670,7 +6736,7 @@
* The name serves as a prefix in the reported metrics names. In case no name has been provided, the default name "reactor" will be applied.
*
* The {@link MeterRegistry} used by reactor can be configured via
- * {@link Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
+ * {@link reactor.util.Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
* prior to using this operator, the default being
* {@link io.micrometer.core.instrument.Metrics#globalRegistry}.
*
@@ -7950,7 +8016,7 @@
* The companion is generated by the provided {@link Retry} instance, see {@link Retry#max(long)}, {@link Retry#maxInARow(long)}
* and {@link Retry#backoff(long, Duration)} for readily available strategy builders.
*
- * The operator generates a base for the companion, a {@link Flux} of {@link Retry.RetrySignal}
+ * The operator generates a base for the companion, a {@link Flux} of {@link reactor.util.retry.Retry.RetrySignal}
* which each give metadata about each retryable failure whenever this {@link Flux} signals an error. The final companion
* should be derived from that base companion and emit data in response to incoming onNext (although it can emit less
* elements, or delay the emissions).
@@ -7960,12 +8026,12 @@
*
*
*
- * Note that the {@link Retry.RetrySignal} state can be
+ * Note that the {@link reactor.util.retry.Retry.RetrySignal} state can be
* transient and change between each source
* {@link org.reactivestreams.Subscriber#onError(Throwable) onError} or
* {@link org.reactivestreams.Subscriber#onNext(Object) onNext}. If processed with a delay,
* this could lead to the represented state being out of sync with the state at which the retry
- * was evaluated. Map it to {@link Retry.RetrySignal#copy()}
+ * was evaluated. Map it to {@link reactor.util.retry.Retry.RetrySignal#copy()}
* right away to mediate this.
*
* Note that if the companion {@link Publisher} created by the {@code whenFactory}
@@ -7990,12 +8056,12 @@
*
*
* @param retrySpec the {@link Retry} strategy that will generate the companion {@link Publisher},
- * given a {@link Flux} that signals each onError as a {@link Retry.RetrySignal}.
+ * given a {@link Flux} that signals each onError as a {@link reactor.util.retry.Retry.RetrySignal}.
*
* @return a {@link Flux} that retries on onError when a companion {@link Publisher} produces an onNext signal
- * @see Retry#max(long)
- * @see Retry#maxInARow(long)
- * @see Retry#backoff(long, Duration)
+ * @see reactor.util.retry.Retry#max(long)
+ * @see reactor.util.retry.Retry#maxInARow(long)
+ * @see reactor.util.retry.Retry#backoff(long, Duration)
*/
public final Flux retryWhen(Retry retrySpec) {
return onAssembly(new FluxRetryWhen<>(this, retrySpec));
@@ -8259,7 +8325,7 @@
/**
* Expect and emit a single item from this {@link Flux} source or signal
- * {@link NoSuchElementException} for an empty source, or
+ * {@link java.util.NoSuchElementException} for an empty source, or
* {@link IndexOutOfBoundsException} for a source with more than one element.
*
*
@@ -8585,7 +8651,7 @@
* Subscribe a {@link Consumer} to this {@link Flux} that will consume all the
* elements in the sequence. It will request an unbounded demand ({@code Long.MAX_VALUE}).
*
- * For a passive version that observe and forward incoming data see {@link #doOnNext(Consumer)}.
+ * For a passive version that observe and forward incoming data see {@link #doOnNext(java.util.function.Consumer)}.
*
* For a version that gives you more control over backpressure and the request, see
* {@link #subscribe(Subscriber)} with a {@link BaseSubscriber}.
@@ -8612,7 +8678,7 @@
* The subscription will request an unbounded demand ({@code Long.MAX_VALUE}).
*
* For a passive version that observe and forward incoming data see
- * {@link #doOnNext(Consumer)} and {@link #doOnError(Consumer)}.
+ * {@link #doOnNext(java.util.function.Consumer)} and {@link #doOnError(java.util.function.Consumer)}.
*
For a version that gives you more control over backpressure and the request, see
* {@link #subscribe(Subscriber)} with a {@link BaseSubscriber}.
*
@@ -8638,8 +8704,8 @@
* elements in the sequence, handle errors and react to completion. The subscription
* will request unbounded demand ({@code Long.MAX_VALUE}).
*
- * For a passive version that observe and forward incoming data see {@link #doOnNext(Consumer)},
- * {@link #doOnError(Consumer)} and {@link #doOnComplete(Runnable)}.
+ * For a passive version that observe and forward incoming data see {@link #doOnNext(java.util.function.Consumer)},
+ * {@link #doOnError(java.util.function.Consumer)} and {@link #doOnComplete(Runnable)}.
*
For a version that gives you more control over backpressure and the request, see
* {@link #subscribe(Subscriber)} with a {@link BaseSubscriber}.
*
@@ -8670,8 +8736,8 @@
* request the adequate amount of data, or request unbounded demand
* {@code Long.MAX_VALUE} if no such consumer is provided.
*
- * For a passive version that observe and forward incoming data see {@link #doOnNext(Consumer)},
- * {@link #doOnError(Consumer)}, {@link #doOnComplete(Runnable)}
+ * For a passive version that observe and forward incoming data see {@link #doOnNext(java.util.function.Consumer)},
+ * {@link #doOnError(java.util.function.Consumer)}, {@link #doOnComplete(Runnable)}
* and {@link #doOnSubscribe(Consumer)}.
*
For a version that gives you more control over backpressure and the request, see
* {@link #subscribe(Subscriber)} with a {@link BaseSubscriber}.
@@ -8710,8 +8776,8 @@
* elements in the sequence, handle errors and react to completion. Additionally, a {@link Context}
* is tied to the subscription. At subscription, an unbounded request is implicitly made.
*
- * For a passive version that observe and forward incoming data see {@link #doOnNext(Consumer)},
- * {@link #doOnError(Consumer)}, {@link #doOnComplete(Runnable)}
+ * For a passive version that observe and forward incoming data see {@link #doOnNext(java.util.function.Consumer)},
+ * {@link #doOnError(java.util.function.Consumer)}, {@link #doOnComplete(Runnable)}
* and {@link #doOnSubscribe(Consumer)}.
*
For a version that gives you more control over backpressure and the request, see
* {@link #subscribe(Subscriber)} with a {@link BaseSubscriber}.
@@ -8770,6 +8836,7 @@
}
}
+ subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(publisher, subscriber);
publisher.subscribe(subscriber);
}
catch (Throwable e) {
@@ -8797,7 +8864,7 @@
* the next occurrence of a {@link #publishOn(Scheduler) publishOn}.
*
* Note that if you are using an eager or blocking
- * {@link #create(Consumer, OverflowStrategy)}
+ * {@link #create(Consumer, FluxSink.OverflowStrategy)}
* as the source, it can lead to deadlocks due to requests piling up behind the emitter.
* In such case, you should call {@link #subscribeOn(Scheduler, boolean) subscribeOn(scheduler, false)}
* instead.
@@ -8834,7 +8901,7 @@
* the next occurrence of a {@link #publishOn(Scheduler) publishOn}.
*
* Note that if you are using an eager or blocking
- * {@link Flux#create(Consumer, OverflowStrategy)}
+ * {@link Flux#create(Consumer, FluxSink.OverflowStrategy)}
* as the source, it can lead to deadlocks due to requests piling up behind the emitter.
* Thus this operator has a {@code requestOnSeparateThread} parameter, which should be
* set to {@code false} in this case.
@@ -9658,7 +9725,7 @@
}
/**
- * Emit a {@link Tuple2} pair of T1 the current clock time in
+ * Emit a {@link reactor.util.function.Tuple2} pair of T1 the current clock time in
* millis (as a {@link Long} measured by the {@link Schedulers#parallel() parallel}
* Scheduler) and T2 the emitted data (as a {@code T}), for each item from this {@link Flux}.
*
@@ -9674,7 +9741,7 @@
}
/**
- * Emit a {@link Tuple2} pair of T1 the current clock time in
+ * Emit a {@link reactor.util.function.Tuple2} pair of T1 the current clock time in
* millis (as a {@link Long} measured by the provided {@link Scheduler}) and T2
* the emitted data (as a {@code T}), for each item from this {@link Flux}.
*
@@ -9916,7 +9983,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -9956,7 +10023,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: The overlapping variant DOES NOT discard elements, as they might be part of another still valid window.
* The exact window and dropping window variants bot discard elements they internally queued for backpressure
@@ -9992,7 +10059,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors and those emitted by the {@code boundary} delivered to the window
- * {@link Flux} are wrapped in {@link Exceptions.SourceException}.
+ * {@link Flux} are wrapped in {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10024,7 +10091,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10067,7 +10134,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: The overlapping variant DOES NOT discard elements, as they might be part of another still valid window.
* The exact window and dropping window variants bot discard elements they internally queued for backpressure
@@ -10101,7 +10168,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10145,7 +10212,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: The overlapping variant DOES NOT discard elements, as they might be part of another still valid window.
* The exact window and dropping window variants bot discard elements they internally queued for backpressure
@@ -10185,7 +10252,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10219,7 +10286,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10255,7 +10322,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10290,7 +10357,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal.
@@ -10330,7 +10397,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10372,7 +10439,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10416,7 +10483,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10455,7 +10522,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10486,7 +10553,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10519,7 +10586,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal. Upon cancellation of the current window,
@@ -10559,7 +10626,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal, as well as the triggering element(s) (that doesn't match
@@ -10597,7 +10664,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator discards elements it internally queued for backpressure
* upon cancellation or error triggered by a data signal, as well as the triggering element(s) (that doesn't match
@@ -10641,7 +10708,7 @@
*
* To distinguish errors emitted by the processing of individual windows, source
* sequence errors delivered to the window {@link Flux} are wrapped in
- * {@link Exceptions.SourceException}.
+ * {@link reactor.core.Exceptions.SourceException}.
*
*
Discard Support: This operator DOES NOT discard elements.
*
@@ -11065,8 +11132,12 @@
*/
@SuppressWarnings("unchecked")
static Flux wrap(Publisher source) {
- if (source instanceof Flux) {
- return (Flux) source;
+ boolean shouldWrap = ContextPropagationSupport.shouldWrapPublisher(source);
+ if (source instanceof Flux) {
+ if (!shouldWrap) {
+ return (Flux) source;
+ }
+ return ContextPropagation.fluxRestoreThreadLocals((Flux) source);
}
//for scalars we'll instantiate the operators directly to avoid onAssembly
@@ -11084,16 +11155,22 @@
}
}
- if(source instanceof Mono){
- if(source instanceof Fuseable){
- return new FluxSourceMonoFuseable<>((Mono)source);
+ Flux target;
+ if (source instanceof Mono) {
+ if (source instanceof Fuseable) {
+ target = new FluxSourceMonoFuseable<>((Mono) source);
+ } else {
+ target = new FluxSourceMono<>((Mono) source);
}
- return new FluxSourceMono<>((Mono)source);
+ } else if (source instanceof Fuseable) {
+ target = new FluxSourceFuseable<>(source);
+ } else {
+ target = new FluxSource<>(source);
}
- if(source instanceof Fuseable){
- return new FluxSourceFuseable<>(source);
+ if (shouldWrap) {
+ return ContextPropagation.fluxRestoreThreadLocals(target);
}
- return new FluxSource<>(source);
+ return target;
}
@SuppressWarnings("rawtypes")
@@ -11108,4 +11185,4 @@
@SuppressWarnings("rawtypes")
static final Function IDENTITY_FUNCTION = Function.identity();
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/FluxArray.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/FluxArray.java (.../FluxArray.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/FluxArray.java (.../FluxArray.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -64,8 +64,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.BUFFERED) return array.length;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
-
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class ArraySubscription
Index: 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferBoundary.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferBoundary.java (.../FluxBufferBoundary.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferBoundary.java (.../FluxBufferBoundary.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -53,7 +53,7 @@
Publisher other,
Supplier bufferSupplier) {
super(source);
- this.other = Objects.requireNonNull(other, "other");
+ this.other = Operators.toFluxOrMono(Objects.requireNonNull(other, "other"));
this.bufferSupplier = Objects.requireNonNull(bufferSupplier, "bufferSupplier");
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferPredicate.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferPredicate.java (.../FluxBufferPredicate.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/FluxBufferPredicate.java (.../FluxBufferPredicate.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -40,7 +40,7 @@
/**
* Buffers elements into custom collections where the buffer boundary is determined by
- * a {@link Predicate} on the values. The predicate can be used in
+ * a {@link java.util.function.Predicate} on the values. The predicate can be used in
* several modes:
*
* For push/pull sinks created using {@link Flux#create(java.util.function.Consumer)}
- * or {@link Flux#create(java.util.function.Consumer, OverflowStrategy)},
+ * or {@link Flux#create(java.util.function.Consumer, FluxSink.OverflowStrategy)},
* the consumer
* is invoked for every request to enable a hybrid backpressure-enabled push/pull model.
*
@@ -118,7 +118,7 @@
* by delivering data to sink only when requests are pending.
*
+ * Unlike in {@link Flux#using(Callable, Function, Consumer) Flux}, in the case of a valued {@link Mono} the cleanup
+ * happens just before passing the value to downstream. In all cases, exceptions raised by the cleanup
+ * {@link Consumer} may override the terminal event, discarding the element if the derived {@link Mono} was valued.
+ *
+ *
+ *
+ * @param resourceSupplier a {@link Callable} that is called on subscribe to create the resource
+ * @param sourceSupplier a {@link Mono} factory to create the Mono depending on the created resource
+ * @param emitted type
+ * @param resource type
+ *
+ * @return new {@link Mono}
+ */
+ public static Mono using(Callable resourceSupplier,
+ Function> sourceSupplier) {
+ return using(resourceSupplier, sourceSupplier, true);
+ }
/**
+ * Uses an {@link AutoCloseable} resource, generated by a supplier for each individual Subscriber,
+ * while streaming the value from a Mono derived from the same resource and makes sure
+ * the resource is released if the sequence terminates or the Subscriber cancels.
+ *
+ *
+ *
For eager cleanup, Unlike in {@link Flux#using(Callable, Function, Consumer) Flux},
+ * in the case of a valued {@link Mono} the cleanup happens just before passing the value to downstream.
+ * In all cases, exceptions raised by the cleanup {@link Consumer} may override the terminal event,
+ * discarding the element if the derived {@link Mono} was valued.
+ *
Non-eager cleanup will drop any exception.
+ *
+ *
+ *
+ *
+ * @param resourceSupplier a {@link Callable} that is called on subscribe to create the resource
+ * @param sourceSupplier a {@link Mono} factory to create the Mono depending on the created resource
+ * @param eager set to true to clean before any signal (including onNext) is passed downstream
+ * @param emitted type
+ * @param resource type
+ *
+ * @return new {@link Mono}
+ */
+ public static Mono using(Callable resourceSupplier,
+ Function> sourceSupplier, boolean eager) {
+ return using(resourceSupplier, sourceSupplier, Exceptions.AUTO_CLOSE, eager);
+ }
+
+ /**
* Uses a resource, generated by a {@link Publisher} for each individual {@link Subscriber},
* to derive a {@link Mono}. Note that all steps of the operator chain that would need the
* resource to be in an open stable state need to be described inside the {@code resourceClosure}
@@ -1717,7 +1784,8 @@
* received or a timeout expires. Returns that value, or null if the Mono completes
* empty. In case the Mono errors, the original exception is thrown (wrapped in a
* {@link RuntimeException} if it was a checked exception).
- * If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
*
*
*
@@ -1726,6 +1794,7 @@
* might miss signal from hot publishers.
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
*
* @return T the result
*/
@@ -1769,7 +1838,8 @@
* Exception via {@link Optional#orElseThrow(Supplier)}.
* In case the Mono itself errors, the original exception is thrown (wrapped in a
* {@link RuntimeException} if it was a checked exception).
- * If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
*
*
*
@@ -1778,6 +1848,7 @@
* might miss signal from hot publishers.
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
*
* @return T the result
*/
@@ -2853,7 +2924,7 @@
}
/**
- * Map this {@link Mono} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Mono} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time between
* the subscribe and the first next signal, as measured by the {@link Schedulers#parallel() parallel} scheduler.
*
@@ -2868,7 +2939,7 @@
}
/**
- * Map this {@link Mono} sequence into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Mono} sequence into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time between the subscribe and the first
* next signal, as measured by the provided {@link Scheduler}.
*
@@ -3478,7 +3549,7 @@
* The name serves as a prefix in the reported metrics names. In case no name has been provided, the default name "reactor" will be applied.
*
* The {@link MeterRegistry} used by reactor can be configured via
- * {@link Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
+ * {@link reactor.util.Metrics.MicrometerConfiguration#useRegistry(MeterRegistry)}
* prior to using this operator, the default being
* {@link io.micrometer.core.instrument.Metrics#globalRegistry}.
*
@@ -4157,7 +4228,7 @@
* The companion is generated by the provided {@link Retry} instance, see {@link Retry#max(long)}, {@link Retry#maxInARow(long)}
* and {@link Retry#backoff(long, Duration)} for readily available strategy builders.
*
- * The operator generates a base for the companion, a {@link Flux} of {@link Retry.RetrySignal}
+ * The operator generates a base for the companion, a {@link Flux} of {@link reactor.util.retry.Retry.RetrySignal}
* which each give metadata about each retryable failure whenever this {@link Mono} signals an error. The final companion
* should be derived from that base companion and emit data in response to incoming onNext (although it can emit less
* elements, or delay the emissions).
@@ -4167,11 +4238,11 @@
*
*
*
- * Note that the {@link Retry.RetrySignal} state can be transient and change between each source
+ * Note that the {@link reactor.util.retry.Retry.RetrySignal} state can be transient and change between each source
* {@link org.reactivestreams.Subscriber#onError(Throwable) onError} or
* {@link org.reactivestreams.Subscriber#onNext(Object) onNext}. If processed with a delay,
* this could lead to the represented state being out of sync with the state at which the retry
- * was evaluated. Map it to {@link Retry.RetrySignal#copy()} right away to mediate this.
+ * was evaluated. Map it to {@link reactor.util.retry.Retry.RetrySignal#copy()} right away to mediate this.
*
* Note that if the companion {@link Publisher} created by the {@code whenFactory}
* emits {@link Context} as trigger objects, these {@link Context} will be merged with
@@ -4196,7 +4267,7 @@
*
*
* @param retrySpec the {@link Retry} strategy that will generate the companion {@link Publisher},
- * given a {@link Flux} that signals each onError as a {@link Retry.RetrySignal}.
+ * given a {@link Flux} that signals each onError as a {@link reactor.util.retry.Retry.RetrySignal}.
*
* @return a {@link Mono} that retries on onError when a companion {@link Publisher} produces an onNext signal
* @see Retry#max(long)
@@ -4210,9 +4281,10 @@
/**
* Prepare a {@link Mono} which shares this {@link Mono} result similar to {@link Flux#shareNext()}.
* This will effectively turn this {@link Mono} into a hot task when the first
- * {@link Subscriber} subscribes using {@link #subscribe()} API. Further {@link Subscriber} will share the same {@link Subscription}
+ * {@link Subscriber} subscribes using {@link #subscribe()} API.
+ * Further {@link Subscriber} will share the same {@link Subscription}
* and therefore the same result.
- * It's worth noting this is an un-cancellable {@link Subscription}.
+ * When all subscribers have cancelled it will cancel the source {@link Mono}.
*
*
*
@@ -4231,7 +4303,7 @@
/**
* Expect exactly one item from this {@link Mono} source or signal
- * {@link NoSuchElementException} for an empty source.
+ * {@link java.util.NoSuchElementException} for an empty source.
*
*
*
@@ -4327,7 +4399,7 @@
* Subscribe a {@link Consumer} to this {@link Mono} that will consume all the
* sequence. It will request an unbounded demand ({@code Long.MAX_VALUE}).
*
- * For a passive version that observe and forward incoming data see {@link #doOnNext(Consumer)}.
+ * For a passive version that observe and forward incoming data see {@link #doOnNext(java.util.function.Consumer)}.
*
* Keep in mind that since the sequence can be asynchronous, this will immediately
* return control to the calling thread. This can give the impression the consumer is
@@ -4351,7 +4423,7 @@
* The subscription will request an unbounded demand ({@code Long.MAX_VALUE}).
*
* For a passive version that observe and forward incoming data see {@link #doOnSuccess(Consumer)} and
- * {@link #doOnError(Consumer)}.
+ * {@link #doOnError(java.util.function.Consumer)}.
*
* Keep in mind that since the sequence can be asynchronous, this will immediately
* return control to the calling thread. This can give the impression the consumer is
@@ -4376,7 +4448,7 @@
* will request unbounded demand ({@code Long.MAX_VALUE}).
*
* For a passive version that observe and forward incoming data see {@link #doOnSuccess(Consumer)} and
- * {@link #doOnError(Consumer)}.
+ * {@link #doOnError(java.util.function.Consumer)}.
*
* Keep in mind that since the sequence can be asynchronous, this will immediately
* return control to the calling thread. This can give the impression the consumer is
@@ -4406,7 +4478,7 @@
* {@code Long.MAX_VALUE} if no such consumer is provided.
*
* For a passive version that observe and forward incoming data see {@link #doOnSuccess(Consumer)} and
- * {@link #doOnError(Consumer)}.
+ * {@link #doOnError(java.util.function.Consumer)}.
*
* Keep in mind that since the sequence can be asynchronous, this will immediately
* return control to the calling thread. This can give the impression the consumer is
@@ -4438,7 +4510,7 @@
* is tied to the subscription. At subscription, an unbounded request is implicitly made.
*
* For a passive version that observe and forward incoming data see {@link #doOnSuccess(Consumer)} and
- * {@link #doOnError(Consumer)}.
+ * {@link #doOnError(java.util.function.Consumer)}.
*
* Keep in mind that since the sequence can be asynchronous, this will immediately
* return control to the calling thread. This can give the impression the consumer is
@@ -4492,6 +4564,7 @@
}
}
+ subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(publisher, subscriber);
publisher.subscribe(subscriber);
}
catch (Throwable e) {
@@ -4521,7 +4594,7 @@
*
*
*
*
* @param scheduler a {@link Scheduler} providing the {@link Worker} where to subscribe
@@ -5024,7 +5097,7 @@
}
/**
- * If this {@link Mono} is valued, emit a {@link Tuple2} pair of
+ * If this {@link Mono} is valued, emit a {@link reactor.util.function.Tuple2} pair of
* T1 the current clock time in millis (as a {@link Long} measured by the
* {@link Schedulers#parallel() parallel} Scheduler) and T2 the emitted data (as a {@code T}).
*
@@ -5039,7 +5112,7 @@
}
/**
- * If this {@link Mono} is valued, emit a {@link Tuple2} pair of
+ * If this {@link Mono} is valued, emit a {@link reactor.util.function.Tuple2} pair of
* T1 the current clock time in millis (as a {@link Long} measured by the
* provided {@link Scheduler}) and T2 the emitted data (as a {@code T}).
*
@@ -5324,52 +5397,67 @@
* Note that this bypasses {@link Hooks#onEachOperator(String, Function) assembly hooks}.
*
* @param source the {@link Publisher} to wrap
- * @param enforceMonoContract {@code} true to wrap publishers without assumption about their cardinality
+ * @param enforceMonoContract {@code true} to wrap publishers without assumption about their cardinality
* (first {@link Subscriber#onNext(Object)} will cancel the source), {@code false} to behave like {@link #fromDirect(Publisher)}.
* @param input upstream type
* @return a wrapped {@link Mono}
*/
static Mono wrap(Publisher source, boolean enforceMonoContract) {
//some sources can be considered already assembled monos
//all conversion methods (from, fromDirect, wrap) must accommodate for this
+ boolean shouldWrap = ContextPropagationSupport.shouldWrapPublisher(source);
if (source instanceof Mono) {
- return (Mono) source;
+ if (!shouldWrap) {
+ return (Mono) source;
+ }
+ return ContextPropagation.monoRestoreThreadLocals((Mono) source);
}
- if (source instanceof FluxSourceMono
- || source instanceof FluxSourceMonoFuseable) {
- @SuppressWarnings("unchecked")
- Mono extracted = (Mono) ((FluxFromMonoOperator) source).source;
- return extracted;
+
+ if (source instanceof FluxSourceMono || source instanceof FluxSourceMonoFuseable) {
+ @SuppressWarnings("unchecked") Mono extracted =
+ (Mono) ((FluxFromMonoOperator) source).source;
+ boolean shouldWrapExtracted =
+ ContextPropagationSupport.shouldWrapPublisher(extracted);
+ if (!shouldWrapExtracted) {
+ return extracted;
+ }
+ return ContextPropagation.monoRestoreThreadLocals(extracted);
}
+ if (source instanceof Flux && source instanceof Callable) {
+ @SuppressWarnings("unchecked") Callable m = (Callable) source;
+ return Flux.wrapToMono(m);
+ }
+
+ Mono target;
+
//equivalent to what from used to be, without assembly hooks
if (enforceMonoContract) {
- if (source instanceof Flux && source instanceof Callable) {
- @SuppressWarnings("unchecked") Callable m = (Callable) source;
- return Flux.wrapToMono(m);
- }
if (source instanceof Flux) {
- return new MonoNext<>((Flux) source);
+ target = new MonoNext<>((Flux) source);
+ } else {
+ target = new MonoFromPublisher<>(source);
}
- return new MonoFromPublisher<>(source);
+ //equivalent to what fromDirect used to be without onAssembly
+ } else if (source instanceof Flux && source instanceof Fuseable) {
+ target = new MonoSourceFluxFuseable<>((Flux) source);
+ } else if (source instanceof Flux) {
+ target = new MonoSourceFlux<>((Flux) source);
+ } else if (source instanceof Fuseable) {
+ target = new MonoSourceFuseable<>(source);
+ } else {
+ target = new MonoSource<>(source);
}
- //equivalent to what fromDirect used to be without onAssembly
- if(source instanceof Flux && source instanceof Fuseable) {
- return new MonoSourceFluxFuseable<>((Flux) source);
+ if (shouldWrap) {
+ return ContextPropagation.monoRestoreThreadLocals(target);
}
- if (source instanceof Flux) {
- return new MonoSourceFlux<>((Flux) source);
- }
- if(source instanceof Fuseable) {
- return new MonoSourceFuseable<>(source);
- }
- return new MonoSource<>(source);
+ return target;
}
@SuppressWarnings("unchecked")
static BiPredicate equalsBiPredicate(){
return EQUALS_BIPREDICATE;
}
static final BiPredicate EQUALS_BIPREDICATE = Object::equals;
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoCacheInvalidateWhen.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoCacheInvalidateWhen.java (.../MonoCacheInvalidateWhen.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoCacheInvalidateWhen.java (.../MonoCacheInvalidateWhen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2021-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -282,6 +282,9 @@
for (@SuppressWarnings("unchecked") CacheMonoSubscriber inner : SUBSCRIBERS.getAndSet(this, COORDINATOR_DONE)) {
inner.complete(value);
}
+ // even though the trigger can deliver values on different threads,
+ // it's not causing any delivery to downstream, so we don't need to
+ // wrap it
invalidateTrigger.subscribe(new TriggerSubscriber(this.main));
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoCallable.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoCallable.java (.../MonoCallable.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoCallable.java (.../MonoCallable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -75,7 +75,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static class MonoCallableSubscription
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoCompletionStage.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoCompletionStage.java (.../MonoCompletionStage.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoCompletionStage.java (.../MonoCompletionStage.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -52,19 +52,15 @@
@Override
public void subscribe(CoreSubscriber actual) {
- if (ContextPropagationSupport.shouldPropagateContextToThreadLocals()) {
- actual.onSubscribe(
- new MonoCompletionStageRestoringThreadLocalsSubscription<>(
- actual, future, suppressCancellation));
- } else {
- actual.onSubscribe(new MonoCompletionStageSubscription<>(
- actual, future, suppressCancellation));
- }
+ CoreSubscriber wrapped = Operators.restoreContextOnSubscriberIfAutoCPEnabled(this, actual);
+ wrapped.onSubscribe(new MonoCompletionStageSubscription<>(
+ wrapped, future, suppressCancellation));
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
@@ -165,108 +161,4 @@
}
}
}
-
- static class MonoCompletionStageRestoringThreadLocalsSubscription
- implements InnerProducer, BiFunction {
-
- final CoreSubscriber actual;
- final CompletionStage future;
- final boolean suppressCancellation;
-
- volatile int requestedOnce;
- @SuppressWarnings("rawtypes")
- static final AtomicIntegerFieldUpdater REQUESTED_ONCE =
- AtomicIntegerFieldUpdater.newUpdater(MonoCompletionStageRestoringThreadLocalsSubscription.class, "requestedOnce");
-
- volatile boolean cancelled;
-
- MonoCompletionStageRestoringThreadLocalsSubscription(
- CoreSubscriber actual,
- CompletionStage future,
- boolean suppressCancellation) {
- this.actual = actual;
- this.future = future;
- this.suppressCancellation = suppressCancellation;
- }
-
- @Override
- public CoreSubscriber actual() {
- return this.actual;
- }
-
- @Override
- public Void apply(@Nullable T value, @Nullable Throwable e) {
- final CoreSubscriber actual = this.actual;
-
- try (ContextSnapshot.Scope ignored =
- ContextPropagation.setThreadLocals(actual.currentContext())) {
- if (this.cancelled) {
- //nobody is interested in the Mono anymore, don't risk dropping errors
- final Context ctx = actual.currentContext();
- if (e == null || e instanceof CancellationException) {
- //we discard any potential value and ignore Future cancellations
- Operators.onDiscard(value, ctx);
- }
- else {
- //we make sure we keep _some_ track of a Future failure AFTER the Mono cancellation
- Operators.onErrorDropped(e, ctx);
- //and we discard any potential value just in case both e and v are not null
- Operators.onDiscard(value, ctx);
- }
-
- return null;
- }
-
- try {
- if (e instanceof CompletionException) {
- actual.onError(e.getCause());
- }
- else if (e != null) {
- actual.onError(e);
- }
- else if (value != null) {
- actual.onNext(value);
- actual.onComplete();
- }
- else {
- actual.onComplete();
- }
- }
- catch (Throwable e1) {
- Operators.onErrorDropped(e1, actual.currentContext());
- throw Exceptions.bubble(e1);
- }
- return null;
- }
- }
-
- @Override
- public void request(long n) {
- if (this.cancelled) {
- return;
- }
-
- if (this.requestedOnce == 1 || !REQUESTED_ONCE.compareAndSet(this, 0 , 1)) {
- return;
- }
-
- future.handle(this);
- }
-
- @Override
- public void cancel() {
- this.cancelled = true;
-
- final CompletionStage future = this.future;
- if (!suppressCancellation && future instanceof Future) {
- try {
- //noinspection unchecked
- ((Future) future).cancel(true);
- }
- catch (Throwable t) {
- Operators.onErrorDropped(t, this.actual.currentContext());
- }
- }
- }
- }
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoContextWriteRestoringThreadLocals.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoContextWriteRestoringThreadLocals.java (.../MonoContextWriteRestoringThreadLocals.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoContextWriteRestoringThreadLocals.java (.../MonoContextWriteRestoringThreadLocals.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -48,6 +48,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return super.scanUnsafe(key);
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoCreate.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoCreate.java (.../MonoCreate.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoCreate.java (.../MonoCreate.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -50,22 +50,26 @@
@Override
public void subscribe(CoreSubscriber actual) {
- DefaultMonoSink emitter = new DefaultMonoSink<>(actual);
+ CoreSubscriber wrapped =
+ Operators.restoreContextOnSubscriberIfAutoCPEnabled(this, actual);
- actual.onSubscribe(emitter);
+ DefaultMonoSink emitter = new DefaultMonoSink<>(wrapped);
+ wrapped.onSubscribe(emitter);
+
try {
callback.accept(emitter);
}
catch (Throwable ex) {
- emitter.error(Operators.onOperatorError(ex, actual.currentContext()));
+ emitter.error(Operators.onOperatorError(ex, wrapped.currentContext()));
}
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
- return null;
+ if (key == InternalProducerAttr.INSTANCE) return true;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class DefaultMonoSink extends AtomicBoolean
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoCurrentContext.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoCurrentContext.java (.../MonoCurrentContext.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoCurrentContext.java (.../MonoCurrentContext.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -39,6 +39,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDefer.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDefer.java (.../MonoDefer.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDefer.java (.../MonoDefer.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -50,12 +50,12 @@
return;
}
- p.subscribe(actual);
+ fromDirect(p).subscribe(actual);
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDeferContextual.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDeferContextual.java (.../MonoDeferContextual.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDeferContextual.java (.../MonoDeferContextual.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -52,12 +52,12 @@
return;
}
- p.subscribe(actual);
+ Operators.toFluxOrMono(p).subscribe(actual);
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null; //no particular key to be represented, still useful in hooks
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDelay.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDelay.java (.../MonoDelay.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDelay.java (.../MonoDelay.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -71,8 +71,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_ON) return timedScheduler;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
-
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class MonoDelayRunnable implements Runnable, InnerProducer {
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDelaySubscription.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDelaySubscription.java (.../MonoDelaySubscription.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDelaySubscription.java (.../MonoDelaySubscription.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -38,7 +38,7 @@
MonoDelaySubscription(Mono source, Publisher other) {
super(source);
- this.other = Objects.requireNonNull(other, "other");
+ this.other = Operators.toFluxOrMono(Objects.requireNonNull(other, "other"));
}
@Override
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDelayUntil.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDelayUntil.java (.../MonoDelayUntil.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDelayUntil.java (.../MonoDelayUntil.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -94,7 +94,7 @@
@Override
public void subscribe(CoreSubscriber actual) {
try {
- source.subscribe(subscribeOrReturn(actual));
+ Operators.toFluxOrMono(source).subscribe(subscribeOrReturn(actual));
}
catch (Throwable e) {
Operators.error(actual, Operators.onOperatorError(e, actual.currentContext()));
@@ -123,6 +123,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null; //no particular key to be represented, still useful in hooks
}
@@ -286,7 +287,7 @@
final Function> generator =
this.otherGenerators[this.index];
- final Publisher p;
+ Publisher p;
try {
p = generator.apply(this.value);
@@ -303,6 +304,7 @@
this.triggerSubscriber = triggerSubscriber;
}
+ p = Operators.toFluxOrMono(p);
p.subscribe(triggerSubscriber);
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoDoOnEachFuseable.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoDoOnEachFuseable.java (.../MonoDoOnEachFuseable.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoDoOnEachFuseable.java (.../MonoDoOnEachFuseable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -23,7 +23,7 @@
import reactor.core.Fuseable;
/**
- * Peek into the lifecycle events and signals of a sequence, {@link Fuseable}
+ * Peek into the lifecycle events and signals of a sequence, {@link reactor.core.Fuseable}
* version of {@link MonoDoOnEach}.
*
* @param the value type
@@ -49,4 +49,4 @@
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
return super.scanUnsafe(key);
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoEmpty.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoEmpty.java (.../MonoEmpty.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoEmpty.java (.../MonoEmpty.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -78,6 +78,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoError.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoError.java (.../MonoError.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoError.java (.../MonoError.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -64,6 +64,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoErrorSupplied.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoErrorSupplied.java (.../MonoErrorSupplied.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoErrorSupplied.java (.../MonoErrorSupplied.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -68,6 +68,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFilterWhen.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFilterWhen.java (.../MonoFilterWhen.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFilterWhen.java (.../MonoFilterWhen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -143,6 +143,7 @@
}
else {
FilterWhenInner inner = new FilterWhenInner<>(this, !(p instanceof Mono), t);
+ p = Operators.toFluxOrMono(p);
p.subscribe(inner);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithSignal.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithSignal.java (.../MonoFirstWithSignal.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithSignal.java (.../MonoFirstWithSignal.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -136,11 +136,14 @@
actual.currentContext()));
}
else {
+ p = Operators.toFluxOrMono(p);
p.subscribe(actual);
}
return;
}
+ Operators.toFluxOrMono(a);
+
FluxFirstWithSignal.RaceCoordinator coordinator =
new FluxFirstWithSignal.RaceCoordinator<>(n);
@@ -150,7 +153,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
\ No newline at end of file
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithValue.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithValue.java (.../MonoFirstWithValue.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFirstWithValue.java (.../MonoFirstWithValue.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -147,7 +147,7 @@
return;
}
if (n == 1) {
- Publisher p = a[0];
+ Publisher p = Mono.from(a[0]);
if (p == null) {
Operators.error(actual,
@@ -169,6 +169,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMap.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMap.java (.../MonoFlatMap.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMap.java (.../MonoFlatMap.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -162,7 +162,7 @@
}
try {
- m.subscribe(new FlatMapInner<>(this));
+ Mono.fromDirect(m).subscribe(new FlatMapInner<>(this));
}
catch (Throwable e) {
actual.onError(Operators.onOperatorError(this, e, t,
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMapMany.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMapMany.java (.../MonoFlatMapMany.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFlatMapMany.java (.../MonoFlatMapMany.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -192,6 +192,7 @@
return;
}
+ p = Operators.toFluxOrMono(p);
p.subscribe(new FlatMapManyInner<>(this, actual));
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFromFluxOperator.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFromFluxOperator.java (.../MonoFromFluxOperator.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFromFluxOperator.java (.../MonoFromFluxOperator.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -62,6 +62,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return Integer.MAX_VALUE;
if (key == Attr.PARENT) return source;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
@@ -78,6 +79,7 @@
}
OptimizableOperator newSource = operator.nextOptimizableSource();
if (newSource == null) {
+ subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(operator.source(), subscriber);
operator.source().subscribe(subscriber);
return;
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoFromPublisher.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoFromPublisher.java (.../MonoFromPublisher.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoFromPublisher.java (.../MonoFromPublisher.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -55,15 +55,12 @@
@Override
@SuppressWarnings("unchecked")
public void subscribe(CoreSubscriber actual) {
- if (ContextPropagationSupport.shouldPropagateContextToThreadLocals()) {
- actual = new MonoSource.MonoSourceRestoringThreadLocalsSubscriber<>(actual);
- }
-
try {
CoreSubscriber subscriber = subscribeOrReturn(actual);
if (subscriber == null) {
return;
}
+ subscriber = Operators.restoreContextOnSubscriberIfPublisherNonInternal(source, subscriber);
source.subscribe(subscriber);
}
catch (Throwable e) {
@@ -89,13 +86,16 @@
@Override
@Nullable
- public Object scanUnsafe(Attr key) {
- if (key == Attr.PARENT) {
+ public Object scanUnsafe(Scannable.Attr key) {
+ if (key == Scannable.Attr.PARENT) {
return source;
}
- if (key == Attr.RUN_STYLE) {
+ if (key == Scannable.Attr.RUN_STYLE) {
return Attr.RunStyle.SYNC;
}
+ if (key == InternalProducerAttr.INSTANCE) {
+ return true;
+ }
return null;
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnorePublisher.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnorePublisher.java (.../MonoIgnorePublisher.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnorePublisher.java (.../MonoIgnorePublisher.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -40,7 +40,7 @@
final OptimizableOperator optimizableOperator;
MonoIgnorePublisher(Publisher source) {
- this.source = Objects.requireNonNull(source, "publisher");
+ this.source = Operators.toFluxOrMono(Objects.requireNonNull(source, "publisher"));
if (source instanceof OptimizableOperator) {
@SuppressWarnings("unchecked")
OptimizableOperator optimSource = (OptimizableOperator) source;
@@ -79,13 +79,16 @@
@Override
@Nullable
- public Object scanUnsafe(Attr key) {
- if (key == Attr.PARENT) {
+ public Object scanUnsafe(Scannable.Attr key) {
+ if (key == Scannable.Attr.PARENT) {
return source;
}
if (key == Attr.RUN_STYLE) {
return Attr.RunStyle.SYNC;
}
+ if (key == InternalProducerAttr.INSTANCE) {
+ return true;
+ }
return null;
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnoreThen.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnoreThen.java (.../MonoIgnoreThen.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoIgnoreThen.java (.../MonoIgnoreThen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -71,6 +71,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
@@ -237,19 +238,19 @@
}
onComplete();
} else {
- m.subscribe(this);
+ Operators.toFluxOrMono(m).subscribe(this);
}
return;
} else {
- final Publisher m = a[i];
+ Publisher p = a[i];
- if (m instanceof Callable) {
+ if (p instanceof Callable) {
if (isCancelled(this.state)) {
//NB: in the non-callable case, this is handled by activeSubscription.cancel()
return;
}
try {
- Operators.onDiscard(((Callable) m).call(), currentContext());
+ Operators.onDiscard(((Callable) p).call(), currentContext());
}
catch (Throwable ex) {
onError(Operators.onOperatorError(ex, currentContext()));
@@ -260,7 +261,8 @@
continue;
}
- m.subscribe((CoreSubscriber) this);
+ p = Operators.toFluxOrMono(p);
+ p.subscribe((CoreSubscriber) this);
return;
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoJust.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoJust.java (.../MonoJust.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoJust.java (.../MonoJust.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2015-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2015-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -59,6 +59,6 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.BUFFERED) return 1;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoLift.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoLift.java (.../MonoLift.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoLift.java (.../MonoLift.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -38,7 +38,7 @@
@Override
public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {
CoreSubscriber input =
- liftFunction.lifter.apply(source, actual);
+ liftFunction.lifter.apply(source, Operators.restoreContextOnSubscriberIfAutoCPEnabled(source, actual));
Objects.requireNonNull(input, "Lifted subscriber MUST NOT be null");
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoLiftFuseable.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoLiftFuseable.java (.../MonoLiftFuseable.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoLiftFuseable.java (.../MonoLiftFuseable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -55,9 +55,8 @@
@Override
public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {
+ CoreSubscriber input = liftFunction.lifter.apply(source, Operators.restoreContextOnSubscriberIfAutoCPEnabled(source, actual));
- CoreSubscriber input = liftFunction.lifter.apply(source, actual);
-
Objects.requireNonNull(input, "Lifted subscriber MUST NOT be null");
if (actual instanceof QueueSubscription
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoMetricsFuseable.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoMetricsFuseable.java (.../MonoMetricsFuseable.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoMetricsFuseable.java (.../MonoMetricsFuseable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -76,7 +76,7 @@
int mode;
@Nullable
- QueueSubscription qs;
+ Fuseable.QueueSubscription qs;
MetricsFuseableSubscriber(CoreSubscriber actual,
MeterRegistry registry,
@@ -196,4 +196,4 @@
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoNever.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoNever.java (.../MonoNever.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoNever.java (.../MonoNever.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -43,7 +43,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
/**
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoOperator.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoOperator.java (.../MonoOperator.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoOperator.java (.../MonoOperator.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,12 +17,8 @@
package reactor.core.publisher;
import java.util.Objects;
-import java.util.function.BiFunction;
-import java.util.function.Function;
-
import org.reactivestreams.Publisher;
-import reactor.core.CoreSubscriber;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
@@ -51,6 +47,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return Integer.MAX_VALUE;
if (key == Attr.PARENT) return source;
+ if (key == InternalProducerAttr.INSTANCE) return false; // public class!
return null;
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoPeekTerminal.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoPeekTerminal.java (.../MonoPeekTerminal.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoPeekTerminal.java (.../MonoPeekTerminal.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -84,7 +84,7 @@
*/
static final class MonoTerminalPeekSubscriber
implements ConditionalSubscriber, InnerOperator,
- QueueSubscription {
+ Fuseable.QueueSubscription {
final CoreSubscriber actual;
final ConditionalSubscriber actualConditional;
@@ -94,7 +94,7 @@
//TODO could go into a common base for all-in-one subscribers? (as well as actual above)
Subscription s;
@Nullable
- QueueSubscription queueSubscription;
+ Fuseable.QueueSubscription queueSubscription;
int sourceMode;
@@ -393,4 +393,5 @@
return queueSubscription == null ? 0 : queueSubscription.size();
}
}
-}
\ No newline at end of file
+}
+
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoPublishMulticast.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoPublishMulticast.java (.../MonoPublishMulticast.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoPublishMulticast.java (.../MonoPublishMulticast.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -53,8 +53,8 @@
public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {
MonoPublishMulticaster multicast = new MonoPublishMulticaster<>(actual.currentContext());
- Mono out = Objects.requireNonNull(transform.apply(fromDirect(multicast)),
- "The transform returned a null Mono");
+ Mono out = fromDirect(
+ Objects.requireNonNull(transform.apply(fromDirect(multicast)), "The transform returned a null Mono"));
if (out instanceof Fuseable) {
out.subscribe(new FluxPublishMulticast.CancelFuseableMulticaster<>(actual, multicast));
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoRetryWhen.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoRetryWhen.java (.../MonoRetryWhen.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoRetryWhen.java (.../MonoRetryWhen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,7 +37,7 @@
final Retry whenSourceFactory;
MonoRetryWhen(Mono source, Retry whenSourceFactory) {
- super(source);
+ super(Mono.fromDirect(source));
this.whenSourceFactory = Objects.requireNonNull(whenSourceFactory, "whenSourceFactory");
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoRunnable.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoRunnable.java (.../MonoRunnable.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoRunnable.java (.../MonoRunnable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2015-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2015-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -77,7 +77,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class MonoRunnableEagerSubscription extends AtomicBoolean implements Subscription {
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSequenceEqual.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSequenceEqual.java (.../MonoSequenceEqual.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSequenceEqual.java (.../MonoSequenceEqual.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -42,8 +42,8 @@
MonoSequenceEqual(Publisher first, Publisher second,
BiPredicate comparer, int prefetch) {
- this.first = Objects.requireNonNull(first, "first");
- this.second = Objects.requireNonNull(second, "second");
+ this.first = Operators.toFluxOrMono(Objects.requireNonNull(first, "first"));
+ this.second = Operators.toFluxOrMono(Objects.requireNonNull(second, "second"));
this.comparer = Objects.requireNonNull(comparer, "comparer");
if(prefetch < 1){
throw new IllegalArgumentException("Buffer size must be strictly positive: " +
@@ -64,7 +64,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return prefetch;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class EqualCoordinator implements InnerProducer {
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleCallable.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleCallable.java (.../MonoSingleCallable.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleCallable.java (.../MonoSingleCallable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -122,6 +122,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptional.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptional.java (.../MonoSingleOptional.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptional.java (.../MonoSingleOptional.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -39,7 +39,7 @@
@Override
public CoreSubscriber subscribeOrReturn(CoreSubscriber> actual) {
- return new SingleOptionalSubscriber<>(actual);
+ return new MonoSingleOptional.SingleOptionalSubscriber<>(actual);
}
@Override
@@ -121,4 +121,4 @@
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptionalCallable.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptionalCallable.java (.../MonoSingleOptionalCallable.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSingleOptionalCallable.java (.../MonoSingleOptionalCallable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -93,6 +93,6 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSource.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSource.java (.../MonoSource.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSource.java (.../MonoSource.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -65,11 +65,7 @@
*/
@Override
public void subscribe(CoreSubscriber actual) {
- if (ContextPropagationSupport.shouldPropagateContextToThreadLocals()) {
- source.subscribe(new MonoSourceRestoringThreadLocalsSubscriber<>(actual));
- } else {
- source.subscribe(actual);
- }
+ source.subscribe(actual);
}
@Override
@@ -96,92 +92,6 @@
if (key == Attr.RUN_STYLE) {
return Scannable.from(source).scanUnsafe(key);
}
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
-
- static final class MonoSourceRestoringThreadLocalsSubscriber
- implements InnerConsumer {
-
- final CoreSubscriber actual;
-
- Subscription s;
- boolean done;
-
- MonoSourceRestoringThreadLocalsSubscriber(CoreSubscriber actual) {
- this.actual = actual;
- }
-
- @Override
- @Nullable
- public Object scanUnsafe(Attr key) {
- if (key == Attr.PARENT) {
- return s;
- }
- if (key == Attr.RUN_STYLE) {
- return Attr.RunStyle.SYNC;
- }
- if (key == Attr.ACTUAL) {
- return actual;
- }
- return null;
- }
-
- @Override
- public Context currentContext() {
- return actual.currentContext();
- }
-
- @SuppressWarnings("try")
- @Override
- public void onSubscribe(Subscription s) {
- // This is needed, as the downstream can then switch threads,
- // continue the subscription using different primitives and omit this operator
- try (ContextSnapshot.Scope ignored =
- ContextPropagation.setThreadLocals(actual.currentContext())) {
- actual.onSubscribe(s);
- }
- }
-
- @SuppressWarnings("try")
- @Override
- public void onNext(T t) {
- this.done = true;
- try (ContextSnapshot.Scope ignored =
- ContextPropagation.setThreadLocals(actual.currentContext())) {
- actual.onNext(t);
- actual.onComplete();
- }
- }
-
- @SuppressWarnings("try")
- @Override
- public void onError(Throwable t) {
- try (ContextSnapshot.Scope ignored =
- ContextPropagation.setThreadLocals(actual.currentContext())) {
- if (this.done) {
- Operators.onErrorDropped(t, actual.currentContext());
- return;
- }
-
- this.done = true;
-
- actual.onError(t);
- }
- }
-
- @SuppressWarnings("try")
- @Override
- public void onComplete() {
- if (this.done) {
- return;
- }
-
- this.done = true;
-
- try (ContextSnapshot.Scope ignored =
- ContextPropagation.setThreadLocals(actual.currentContext())) {
- actual.onComplete();
- }
- }
- }
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoStreamCollector.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoStreamCollector.java (.../MonoStreamCollector.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoStreamCollector.java (.../MonoStreamCollector.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -28,7 +28,7 @@
import reactor.util.context.Context;
/**
- * Collects the values from the source sequence into a {@link Collector}
+ * Collects the values from the source sequence into a {@link java.util.stream.Collector}
* instance.
*
* @param the source value type
@@ -202,4 +202,4 @@
return r;
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOn.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOn.java (.../MonoSubscribeOn.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOn.java (.../MonoSubscribeOn.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -45,7 +45,7 @@
@Override
public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {
- Worker worker = scheduler.createWorker();
+ Scheduler.Worker worker = scheduler.createWorker();
SubscribeOnSubscriber parent = new SubscribeOnSubscriber<>(source,
actual, worker);
@@ -78,7 +78,7 @@
final Publisher parent;
- final Worker worker;
+ final Scheduler.Worker worker;
volatile Subscription s;
@SuppressWarnings("rawtypes")
@@ -211,4 +211,4 @@
worker.dispose();
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnCallable.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnCallable.java (.../MonoSubscribeOnCallable.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnCallable.java (.../MonoSubscribeOnCallable.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -59,10 +59,11 @@
}
@Override
- public Object scanUnsafe(Attr key) {
- if (key == Attr.RUN_ON) return scheduler;
+ public Object scanUnsafe(Scannable.Attr key) {
+ if (key == Scannable.Attr.RUN_ON) return scheduler;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnValue.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnValue.java (.../MonoSubscribeOnValue.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSubscribeOnValue.java (.../MonoSubscribeOnValue.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -67,6 +67,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_ON) return scheduler;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.ASYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return null;
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSupplier.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSupplier.java (.../MonoSupplier.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSupplier.java (.../MonoSupplier.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -70,7 +70,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static class MonoSupplierSubscription
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoSwitchIfEmpty.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoSwitchIfEmpty.java (.../MonoSwitchIfEmpty.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoSwitchIfEmpty.java (.../MonoSwitchIfEmpty.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -32,7 +32,7 @@
MonoSwitchIfEmpty(Mono source, Mono other) {
super(source);
- this.other = Objects.requireNonNull(other, "other");
+ this.other = fromDirect(Objects.requireNonNull(other, "other"));
}
@Override
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoTakeUntilOther.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoTakeUntilOther.java (.../MonoTakeUntilOther.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoTakeUntilOther.java (.../MonoTakeUntilOther.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -35,7 +35,7 @@
MonoTakeUntilOther(Mono source, Publisher other) {
super(source);
- this.other = Objects.requireNonNull(other, "other");
+ this.other = Operators.toFluxOrMono(Objects.requireNonNull(other, "other"));
}
@Override
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoTapRestoringThreadLocals.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoTapRestoringThreadLocals.java (.../MonoTapRestoringThreadLocals.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoTapRestoringThreadLocals.java (.../MonoTapRestoringThreadLocals.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -83,6 +83,7 @@
public Object scanUnsafe(Attr key) {
if (key == Attr.PREFETCH) return -1;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+ if (key == InternalProducerAttr.INSTANCE) return true;
return super.scanUnsafe(key);
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoTimeout.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoTimeout.java (.../MonoTimeout.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoTimeout.java (.../MonoTimeout.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -48,7 +48,7 @@
Publisher firstTimeout,
String timeoutDescription) {
super(source);
- this.firstTimeout = Objects.requireNonNull(firstTimeout, "firstTimeout");
+ this.firstTimeout = Mono.fromDirect(Objects.requireNonNull(firstTimeout, "firstTimeout"));
this.other = null;
this.timeoutDescription = timeoutDescription;
}
@@ -57,8 +57,8 @@
Publisher firstTimeout,
Publisher other) {
super(source);
- this.firstTimeout = Objects.requireNonNull(firstTimeout, "firstTimeout");
- this.other = Objects.requireNonNull(other, "other");
+ this.firstTimeout = Mono.fromDirect(Objects.requireNonNull(firstTimeout, "firstTimeout"));
+ this.other = Mono.fromDirect(Objects.requireNonNull(other, "other"));
this.timeoutDescription = null;
}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoUsing.java
===================================================================
diff -u -r03ee7b3a8cdecd210e54619377d06f9c7cb4014b -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoUsing.java (.../MonoUsing.java) (revision 03ee7b3a8cdecd210e54619377d06f9c7cb4014b)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoUsing.java (.../MonoUsing.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -82,8 +82,8 @@
Mono p;
try {
- p = Objects.requireNonNull(sourceFactory.apply(resource),
- "The sourceFactory returned a null value");
+ p = Mono.fromDirect(Objects.requireNonNull(sourceFactory.apply(resource),
+ "The sourceFactory returned a null value"));
}
catch (Throwable e) {
@@ -117,7 +117,7 @@
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class MonoUsingSubscriber
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoUsingWhen.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoUsingWhen.java (.../MonoUsingWhen.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoUsingWhen.java (.../MonoUsingWhen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2018-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2018-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -84,7 +84,7 @@
asyncCancel,
null);
- p.subscribe(subscriber);
+ fromDirect(p).subscribe(subscriber);
}
}
catch (Throwable e) {
@@ -93,15 +93,15 @@
return;
}
- resourceSupplier.subscribe(new ResourceSubscriber(actual, resourceClosure,
- asyncComplete, asyncError, asyncCancel,
- resourceSupplier instanceof Mono));
+ // Ensure onLastOperatorHook is called by invoking Publisher::subscribe(Subscriber)
+ ((Publisher) Operators.toFluxOrMono(resourceSupplier)).subscribe(
+ new ResourceSubscriber(actual, resourceClosure, asyncComplete, asyncError, asyncCancel, resourceSupplier instanceof Mono));
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
private static Mono deriveMonoFromResource(
@@ -180,7 +180,7 @@
final Mono p = deriveMonoFromResource(resource, resourceClosure);
- p.subscribe(MonoUsingWhen.prepareSubscriberForResource(resource,
+ fromDirect(p).subscribe(MonoUsingWhen.prepareSubscriberForResource(resource,
this.actual,
this.asyncComplete,
this.asyncError,
@@ -243,7 +243,7 @@
}
}
- static class MonoUsingWhenSubscriber extends UsingWhenSubscriber {
+ static class MonoUsingWhenSubscriber extends FluxUsingWhen.UsingWhenSubscriber {
MonoUsingWhenSubscriber(CoreSubscriber actual,
S resource,
@@ -277,4 +277,4 @@
this.actual.onError(error);
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoWhen.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoWhen.java (.../MonoWhen.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoWhen.java (.../MonoWhen.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2017-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2017-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -95,6 +95,8 @@
return;
}
+ Operators.toFluxOrMono(a);
+
WhenCoordinator parent = new WhenCoordinator(a, actual, n, delayError);
actual.onSubscribe(parent);
}
@@ -104,7 +106,7 @@
if (key == Attr.DELAY_ERROR) return delayError;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class WhenCoordinator implements InnerProducer,
Index: 3rdParty_sources/reactor/reactor/core/publisher/MonoZip.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/MonoZip.java (.../MonoZip.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/MonoZip.java (.../MonoZip.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -122,14 +122,18 @@
return;
}
+ for (int i = 0; i < n; i++) {
+ a[i] = Mono.fromDirect(a[i]);
+ }
+
actual.onSubscribe(new ZipCoordinator<>(a, actual, n, delayError, zipper));
}
@Override
public Object scanUnsafe(Attr key) {
if (key == Attr.DELAY_ERROR) return delayError;
if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
- return null;
+ return SourceProducer.super.scanUnsafe(key);
}
static final class ZipCoordinator implements InnerProducer,
Index: 3rdParty_sources/reactor/reactor/core/publisher/NextProcessor.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/NextProcessor.java (.../NextProcessor.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/NextProcessor.java (.../NextProcessor.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -207,9 +207,9 @@
@SuppressWarnings("unused") EmitResult emitResult = tryEmitValue(null);
}
- void emitEmpty(EmitFailureHandler failureHandler) {
+ void emitEmpty(Sinks.EmitFailureHandler failureHandler) {
for (;;) {
- EmitResult emitResult = tryEmitValue(null);
+ Sinks.EmitResult emitResult = tryEmitValue(null);
if (emitResult.isSuccess()) {
return;
}
@@ -269,7 +269,7 @@
}
@SuppressWarnings("unchecked")
- EmitResult tryEmitError(Throwable cause) {
+ Sinks.EmitResult tryEmitError(Throwable cause) {
Objects.requireNonNull(cause, "onError cannot be null");
if (UPSTREAM.getAndSet(this, Operators.cancelledSubscription()) == Operators.cancelledSubscription()) {
@@ -628,4 +628,4 @@
return super.scanUnsafe(key);
}
}
-}
\ No newline at end of file
+}
Index: 3rdParty_sources/reactor/reactor/core/publisher/Operators.java
===================================================================
diff -u -rc4ce08dc0aae7d9da822088a3d5710484f6b0402 -r683b1a0cbc595fe30e200900d081c88105aae63c
--- 3rdParty_sources/reactor/reactor/core/publisher/Operators.java (.../Operators.java) (revision c4ce08dc0aae7d9da822088a3d5710484f6b0402)
+++ 3rdParty_sources/reactor/reactor/core/publisher/Operators.java (.../Operators.java) (revision 683b1a0cbc595fe30e200900d081c88105aae63c)
@@ -1,5 +1,5 @@
/*
- * Copyright (c) 2016-2022 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2023 VMware Inc. or its affiliates, All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -582,7 +582,7 @@
}
/**
- * Invoke a (local or global) hook that processes elements that remains in an {@link Iterator}.
+ * Invoke a (local or global) hook that processes elements that remains in an {@link java.util.Iterator}.
* Since iterators can be infinite, this method requires that you explicitly ensure the iterator is
* {@code knownToBeFinite}. Typically, operating on an {@link Iterable} one can get such a
* guarantee by looking at the {@link Iterable#spliterator() Spliterator's} {@link Spliterator#getExactSizeIfKnown()}.
@@ -620,7 +620,7 @@
/**
- * Invoke a (local or global) hook that processes elements that remains in an {@link Spliterator}.
+ * Invoke a (local or global) hook that processes elements that remains in an {@link java.util.Spliterator}.
* Since spliterators can be infinite, this method requires that you explicitly ensure the spliterator is
* {@code knownToBeFinite}. Typically, one can get such a guarantee by looking at the {@link Spliterator#getExactSizeIfKnown()}.
*
@@ -991,6 +991,57 @@
}
}
+ public static CorePublisher toFluxOrMono(Publisher publisher) {
+ if (publisher instanceof Mono) {
+ return Mono.fromDirect(publisher);
+ }
+ return Flux.from(publisher);
+ }
+
+ public static void toFluxOrMono(Publisher[] sources) {
+ for (int i = 0; i < sources.length; i++) {
+ if (sources[i] != null) {
+ sources[i] = toFluxOrMono(sources[i]);
+ }
+ }
+ }
+
+ static CoreSubscriber restoreContextOnSubscriberIfPublisherNonInternal(
+ Publisher publisher, CoreSubscriber subscriber) {
+ if (ContextPropagationSupport.shouldWrapPublisher(publisher)) {
+ return restoreContextOnSubscriber(publisher, subscriber);
+ }
+ return subscriber;
+ }
+
+ static CoreSubscriber restoreContextOnSubscriberIfAutoCPEnabled(
+ Publisher publisher, CoreSubscriber subscriber) {
+ if (ContextPropagationSupport.shouldPropagateContextToThreadLocals()) {
+ return restoreContextOnSubscriber(publisher, subscriber);
+ }
+ return subscriber;
+ }
+
+ static CoreSubscriber restoreContextOnSubscriber(Publisher publisher, CoreSubscriber subscriber) {
+ if (publisher instanceof Fuseable) {
+ return new FluxContextWriteRestoringThreadLocals.FuseableContextWriteRestoringThreadLocalsSubscriber<>(
+ subscriber, subscriber.currentContext());
+ } else {
+ return new FluxContextWriteRestoringThreadLocals.ContextWriteRestoringThreadLocalsSubscriber<>(
+ subscriber,
+ subscriber.currentContext());
+ }
+ }
+
+ static CoreSubscriber[] restoreContextOnSubscribers(
+ Publisher publisher, CoreSubscriber[] subscribers) {
+ CoreSubscriber[] actualSubscribers = new CoreSubscriber[subscribers.length];
+ for (int i = 0; i < subscribers.length; i++) {
+ actualSubscribers[i] = restoreContextOnSubscriber(publisher, subscribers[i]);
+ }
+ return actualSubscribers;
+ }
+
private static Throwable unwrapOnNextError(Throwable error) {
return Exceptions.isBubbling(error) ? error : Exceptions.unwrap(error);
}
@@ -1341,16 +1392,16 @@
}
/**
- * If the actual {@link CoreSubscriber} is not {@link Fuseable.ConditionalSubscriber},
+ * If the actual {@link CoreSubscriber} is not {@link reactor.core.Fuseable.ConditionalSubscriber},
* it will apply an adapter which directly maps all
- * {@link Fuseable.ConditionalSubscriber#tryOnNext(Object)} to
+ * {@link reactor.core.Fuseable.ConditionalSubscriber#tryOnNext(Object)} to
* {@link Subscriber#onNext(Object)}
* and always returns true as the result
*
* @param passed subscriber type
*
* @param actual the {@link Subscriber} to adapt
- * @return a potentially adapted {@link Fuseable.ConditionalSubscriber}
+ * @return a potentially adapted {@link reactor.core.Fuseable.ConditionalSubscriber}
*/
@SuppressWarnings("unchecked")
public static Fuseable.ConditionalSubscriber toConditionalSubscriber(CoreSubscriber actual) {
@@ -1476,24 +1527,14 @@
Operators() {
}
- static final class CorePublisherAdapter implements CorePublisher,
- OptimizableOperator {
+ static final class CorePublisherAdapter implements CorePublisher {
final Publisher publisher;
- @Nullable
- final OptimizableOperator optimizableOperator;
-
CorePublisherAdapter(Publisher publisher) {
this.publisher = publisher;
- if (publisher instanceof OptimizableOperator) {
- @SuppressWarnings("unchecked")
- OptimizableOperator optimSource = (OptimizableOperator) publisher;
- this.optimizableOperator = optimSource;
- }
- else {
- this.optimizableOperator = null;
- }
+ // note: if publisher is not CorePublisher it can't be an
+ // OptimizableOperator, which extends CorePublisher
}
@Override
@@ -1505,21 +1546,6 @@
public void subscribe(Subscriber s) {
publisher.subscribe(s);
}
-
- @Override
- public CoreSubscriber subscribeOrReturn(CoreSubscriber actual) {
- return actual;
- }
-
- @Override
- public final CorePublisher source() {
- return this;
- }
-
- @Override
- public final OptimizableOperator nextOptimizableSource() {
- return optimizableOperator;
- }
}
static final Fuseable.ConditionalSubscriber EMPTY_SUBSCRIBER = new Fuseable.ConditionalSubscriber
* 
+ * 
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
* @return the first value or null
*/
@Nullable
@@ -2759,7 +2823,8 @@
* signals its last value, completes or a timeout expires. Returns that value,
* or null if the Flux completes empty. In case the Flux errors, the original
* exception is thrown (wrapped in a {@link RuntimeException} if it was a checked
- * exception). If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * exception). If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
* 
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
* @return the last value or null
*/
@Nullable
@@ -5054,7 +5120,7 @@
}
/**
- * Map this {@link Flux} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Flux} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time
* between each signal as measured by the {@link Schedulers#parallel() parallel} scheduler.
* First duration is measured between the subscription and the first element.
@@ -5071,7 +5137,7 @@
}
/**
- * Map this {@link Flux} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Flux} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time
* between each signal as measured by the provided {@link Scheduler}.
* First duration is measured between the subscription and the first element.
@@ -6670,7 +6736,7 @@
* The name serves as a prefix in the reported metrics names. In case no name has been provided, the default name "reactor" will be applied.
* 
* 
+ *
+ * @param resourceSupplier a {@link Callable} that is called on subscribe to create the resource
+ * @param sourceSupplier a {@link Mono} factory to create the Mono depending on the created resource
+ * @param 
@@ -1726,6 +1794,7 @@
* might miss signal from hot publishers.
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
*
* @return T the result
*/
@@ -1769,7 +1838,8 @@
* Exception via {@link Optional#orElseThrow(Supplier)}.
* In case the Mono itself errors, the original exception is thrown (wrapped in a
* {@link RuntimeException} if it was a checked exception).
- * If the provided timeout expires, a {@link RuntimeException} is thrown.
+ * If the provided timeout expires, a {@link RuntimeException} is thrown
+ * with a {@link TimeoutException} as the cause.
*
* 
@@ -1778,6 +1848,7 @@
* might miss signal from hot publishers.
*
* @param timeout maximum time period to wait for before raising a {@link RuntimeException}
+ * with a {@link TimeoutException} as the cause
*
* @return T the result
*/
@@ -2853,7 +2924,7 @@
}
/**
- * Map this {@link Mono} into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Mono} into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time between
* the subscribe and the first next signal, as measured by the {@link Schedulers#parallel() parallel} scheduler.
*
@@ -2868,7 +2939,7 @@
}
/**
- * Map this {@link Mono} sequence into {@link Tuple2 Tuple2<Long, T>}
+ * Map this {@link Mono} sequence into {@link reactor.util.function.Tuple2 Tuple2<Long, T>}
* of timemillis and source data. The timemillis corresponds to the elapsed time between the subscribe and the first
* next signal, as measured by the provided {@link Scheduler}.
*
@@ -3478,7 +3549,7 @@
* The name serves as a prefix in the reported metrics names. In case no name has been provided, the default name "reactor" will be applied.
* 
* 
*
@@ -4231,7 +4303,7 @@
/**
* Expect exactly one item from this {@link Mono} source or signal
- * {@link NoSuchElementException} for an empty source.
+ * {@link java.util.NoSuchElementException} for an empty source.
* 
* 
*
*