https://github.com/reactor/reactor-core: Non-Blocking Reactive Foundation for the JVM (reactor)
  • Stephane Maldini
  • David Karnok
  • Simon Baslé
  • Sergei Egorov
  • Sébastien Deleuze 
/*
 * Copyright (c) 2011-2017 Pivotal Software Inc, 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.
 * You may obtain a copy of the License at
 *
 *       https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package reactor.core.publisher;

import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Stream;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.CoreSubscriber;
import reactor.core.Exceptions;
import reactor.core.Fuseable;
import reactor.core.Scannable;
import reactor.util.annotation.Nullable;
import reactor.util.context.Context;


Maps a sequence of values each into a Publisher and flattens them
back into a single sequence, interleaving events from the various inner Publishers.

Type parameters:
  • <T> – the source value type
  • <R> – the result value type
See Also:
/**
 * Maps a sequence of values each into a Publisher and flattens them
 * back into a single sequence, interleaving events from the various inner Publishers.
 *
 * @param <T> the source value type
 * @param <R> the result value type
 * @see <a href="https://github.com/reactor/reactive-streams-commons">Reactive-Streams-Commons</a>
 */
final class FluxFlatMap<T, R> extends InternalFluxOperator<T, R> {

	final Function<? super T, ? extends Publisher<? extends R>> mapper;

	final boolean delayError;

	final int maxConcurrency;

	final Supplier<? extends Queue<R>> mainQueueSupplier;

	final int prefetch;

	final Supplier<? extends Queue<R>> innerQueueSupplier;

	FluxFlatMap(Flux<? extends T> source,
			Function<? super T, ? extends Publisher<? extends R>> mapper,
			boolean delayError,
			int maxConcurrency,
			Supplier<? extends Queue<R>> mainQueueSupplier,
			int prefetch,
			Supplier<? extends Queue<R>> innerQueueSupplier) {
		super(source);
		if (prefetch <= 0) {
			throw new IllegalArgumentException("prefetch > 0 required but it was " + prefetch);
		}
		if (maxConcurrency <= 0) {
			throw new IllegalArgumentException("maxConcurrency > 0 required but it was " + maxConcurrency);
		}
		this.mapper = Objects.requireNonNull(mapper, "mapper");
		this.delayError = delayError;
		this.prefetch = prefetch;
		this.maxConcurrency = maxConcurrency;
		this.mainQueueSupplier =
				Objects.requireNonNull(mainQueueSupplier, "mainQueueSupplier");
		this.innerQueueSupplier =
				Objects.requireNonNull(innerQueueSupplier, "innerQueueSupplier");
	}

	@Override
	public int getPrefetch() {
		return prefetch;
	}

	@Override
	public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super R> actual) {
		if (trySubscribeScalarMap(source, actual, mapper, false, true)) {
			return null;
		}

		return new FlatMapMain<>(actual,
				mapper,
				delayError,
				maxConcurrency,
				mainQueueSupplier,
				prefetch, innerQueueSupplier);
	}

	
Checks if the source is a Supplier and if the mapper's publisher output is also
a supplier, thus avoiding subscribing to any of them.

Params:
  • source – the source publisher
  • s – the end consumer
  • mapper – the mapper function
  • fuseableExpected – if true, the parent class was marked Fuseable thus the
mapping output has to signal onSubscribe with a QueueSubscription
Returns:true if the optimization worked
/**
	 * Checks if the source is a Supplier and if the mapper's publisher output is also
	 * a supplier, thus avoiding subscribing to any of them.
	 *
	 * @param source the source publisher
	 * @param s the end consumer
	 * @param mapper the mapper function
	 * @param fuseableExpected if true, the parent class was marked Fuseable thus the
	 * mapping output has to signal onSubscribe with a QueueSubscription
	 *
	 * @return true if the optimization worked
	 */
	@SuppressWarnings("unchecked")
	static <T, R> boolean trySubscribeScalarMap(Publisher<? extends T> source,
			CoreSubscriber<? super R> s,
			Function<? super T, ? extends Publisher<? extends R>> mapper,
			boolean fuseableExpected,
			boolean errorContinueExpected) {
		if (source instanceof Callable) {
			T t;

			try {
				t = ((Callable<? extends T>) source).call();
			}
			catch (Throwable e) {
				Context ctx = s.currentContext();
				Throwable e_ = errorContinueExpected ?
					Operators.onNextError(null, e, ctx) :
					Operators.onOperatorError(e, ctx);
				if (e_ != null) {
					Operators.error(s, e_);
				}
				else {
					//the error was recovered but we know there won't be any more value
					Operators.complete(s);
				}
				return true;
			}

			if (t == null) {
				Operators.complete(s);
				return true;
			}

			Publisher<? extends R> p;

			try {
				p = Objects.requireNonNull(mapper.apply(t),
						"The mapper returned a null Publisher");
			}
			catch (Throwable e) {
				Context ctx = s.currentContext();
				Throwable e_ = errorContinueExpected ?
						Operators.onNextError(t, e, ctx) :
						Operators.onOperatorError(null, e, t, ctx);
				if (e_ != null) {
					Operators.error(s, e_);
				}
				else {
					//the error was recovered but we know there won't be any more value
					Operators.complete(s);
				}
				return true;
			}

			if (p instanceof Callable) {
				R v;

				try {
					v = ((Callable<R>) p).call();
				}
				catch (Throwable e) {
					Context ctx = s.currentContext();
					Throwable e_ = errorContinueExpected ?
							Operators.onNextError(t, e, ctx) :
							Operators.onOperatorError(null, e, t, ctx);
					if (e_ != null) {
						Operators.error(s, e_);
					}
					else {
						//the error was recovered but we know there won't be any more value
						Operators.complete(s);
					}
					return true;
				}

				if (v != null) {
					s.onSubscribe(Operators.scalarSubscription(s, v));
				}
				else {
					Operators.complete(s);
				}
			}
			else {
				if (!fuseableExpected || p instanceof Fuseable) {
					p.subscribe(s);
				}
				else {
					p.subscribe(new FluxHide.SuppressFuseableSubscriber<>(s));
				}
			}

			return true;
		}

		return false;
	}

	@Override
	public Object scanUnsafe(Attr key) {
		if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
		return super.scanUnsafe(key);
	}

	static final class FlatMapMain<T, R> extends FlatMapTracker<FlatMapInner<R>>
			implements InnerOperator<T, R> {

		final boolean                                               delayError;
		final int                                                   maxConcurrency;
		final int                                                   prefetch;
		final int                                                   limit;
		final Function<? super T, ? extends Publisher<? extends R>> mapper;
		final Supplier<? extends Queue<R>>                          mainQueueSupplier;
		final Supplier<? extends Queue<R>>                          innerQueueSupplier;
		final CoreSubscriber<? super R>                             actual;

		volatile Queue<R> scalarQueue;

		volatile Throwable error;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<FlatMapMain, Throwable> ERROR =
				AtomicReferenceFieldUpdater.newUpdater(FlatMapMain.class,
						Throwable.class,
						"error");

		volatile boolean done;

		volatile boolean cancelled;

		Subscription s;

		volatile long requested;
		@SuppressWarnings("rawtypes")
		static final AtomicLongFieldUpdater<FlatMapMain> REQUESTED =
				AtomicLongFieldUpdater.newUpdater(FlatMapMain.class, "requested");

		volatile int wip;
		@SuppressWarnings("rawtypes")
		static final AtomicIntegerFieldUpdater<FlatMapMain> WIP =
				AtomicIntegerFieldUpdater.newUpdater(FlatMapMain.class, "wip");

		@SuppressWarnings("rawtypes")
		static final FlatMapInner[] EMPTY = new FlatMapInner[0];

		@SuppressWarnings("rawtypes")
		static final FlatMapInner[] TERMINATED = new FlatMapInner[0];


		int lastIndex;

		int produced;

		FlatMapMain(CoreSubscriber<? super R> actual,
				Function<? super T, ? extends Publisher<? extends R>> mapper,
				boolean delayError,
				int maxConcurrency,
				Supplier<? extends Queue<R>> mainQueueSupplier,
				int prefetch,
				Supplier<? extends Queue<R>> innerQueueSupplier) {
			this.actual = actual;
			this.mapper = mapper;
			this.delayError = delayError;
			this.maxConcurrency = maxConcurrency;
			this.mainQueueSupplier = mainQueueSupplier;
			this.prefetch = prefetch;
			this.innerQueueSupplier = innerQueueSupplier;
			this.limit = Operators.unboundedOrLimit(maxConcurrency);
		}

		@Override
		public final CoreSubscriber<? super R> actual() {
			return actual;
		}

		@Override
		public Stream<? extends Scannable> inners() {
			return Stream.of(array).filter(Objects::nonNull);
		}

		@Override
		@Nullable
		public Object scanUnsafe(Attr key) {
			if (key == Attr.PARENT) return s;
			if (key == Attr.CANCELLED) return cancelled;
			if (key == Attr.ERROR) return error;
			if (key == Attr.TERMINATED) return done && (scalarQueue == null || scalarQueue.isEmpty());
			if (key == Attr.DELAY_ERROR) return delayError;
			if (key == Attr.PREFETCH) return maxConcurrency;
			if (key == Attr.REQUESTED_FROM_DOWNSTREAM) return requested;
			if (key == Attr.LARGE_BUFFERED) return (scalarQueue != null ? (long) scalarQueue.size() : 0L) + size;
			if (key == Attr.BUFFERED) {
				long realBuffered = (scalarQueue != null ? (long) scalarQueue.size() : 0L) + size;
				if (realBuffered <= Integer.MAX_VALUE) return (int) realBuffered;
				return Integer.MIN_VALUE;
			}
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;

			return InnerOperator.super.scanUnsafe(key);
		}

		@SuppressWarnings("unchecked")
		@Override
		FlatMapInner<R>[] empty() {
			return EMPTY;
		}

		@SuppressWarnings("unchecked")
		@Override
		FlatMapInner<R>[] terminated() {
			return TERMINATED;
		}

		@SuppressWarnings("unchecked")
		@Override
		FlatMapInner<R>[] newArray(int size) {
			return new FlatMapInner[size];
		}

		@Override
		void setIndex(FlatMapInner<R> entry, int index) {
			entry.index = index;
		}

		@Override
		void unsubscribeEntry(FlatMapInner<R> entry) {
			entry.cancel();
		}

		@Override
		public void request(long n) {
			if (Operators.validate(n)) {
				Operators.addCap(REQUESTED, this, n);
				drain(null);
			}
		}

		@Override
		public void cancel() {
			if (!cancelled) {
				cancelled = true;

				if (WIP.getAndIncrement(this) == 0) {
					Operators.onDiscardQueueWithClear(scalarQueue, actual.currentContext(), null);
					scalarQueue = null;
					s.cancel();
					unsubscribe();
				}
			}
		}

		@Override
		public void onSubscribe(Subscription s) {
			if (Operators.validate(this.s, s)) {
				this.s = s;

				actual.onSubscribe(this);
				s.request(Operators.unboundedOrPrefetch(maxConcurrency));
			}
		}

		@SuppressWarnings("unchecked")
		@Override
		public void onNext(T t) {
			if (done) {
				Operators.onNextDropped(t, actual.currentContext());
				return;
			}

			Publisher<? extends R> p;

			try {
				p = Objects.requireNonNull(mapper.apply(t),
				"The mapper returned a null Publisher");
			}
			catch (Throwable e) {
				Context ctx = actual.currentContext();
				Throwable e_ = Operators.onNextError(t, e, ctx, s);
				Operators.onDiscard(t, ctx);
				if (e_ != null) {
					onError(e_);
				}
				else {
					tryEmitScalar(null);
				}
				return;
			}

			if (p instanceof Callable) {
				R v;
				try {
					v = ((Callable<R>) p).call();
				}
				catch (Throwable e) {
					Context ctx = actual.currentContext();
					//does the strategy apply? if so, short-circuit the delayError. In any case, don't cancel
					Throwable e_ = Operators.onNextError(t, e, ctx);
					if (e_ == null) {
						tryEmitScalar(null);
					}
					else if (!delayError || !Exceptions.addThrowable(ERROR, this, e_)) {
					//now if error mode strategy doesn't apply, let delayError play
						onError(Operators.onOperatorError(s, e_, t, ctx));
					}
					Operators.onDiscard(t, ctx);
					return;
				}
				tryEmitScalar(v);
			}
			else {
				FlatMapInner<R> inner = new FlatMapInner<>(this, prefetch);
				if (add(inner)) {
					p.subscribe(inner);
				} else {
					Operators.onDiscard(t, actual.currentContext());
				}
			}

		}

		Queue<R> getOrCreateScalarQueue() {
			Queue<R> q = scalarQueue;
			if (q == null) {
				q = mainQueueSupplier.get();
				scalarQueue = q;
			}
			return q;
		}

		@Override
		public void onError(Throwable t) {
			if (done) {
				Operators.onErrorDropped(t, actual.currentContext());
				return;
			}
			if (Exceptions.addThrowable(ERROR, this, t)) {
				done = true;
				drain(null);
			}
			else {
				Operators.onErrorDropped(t, actual.currentContext());
			}
		}

		@Override
		public void onComplete() {
			if (done) {
				return;
			}

			done = true;
			drain(null);
		}

		void tryEmitScalar(@Nullable R v) {
			if (v == null) {
				if (maxConcurrency != Integer.MAX_VALUE) {
					int p = produced + 1;
					if (p == limit) {
						produced = 0;
						s.request(p);
					}
					else {
						produced = p;
					}
				}
				return;
			}

			if (wip == 0 && WIP.compareAndSet(this, 0, 1)) {
				long r = requested;

				Queue<R> q = scalarQueue;
				if (r != 0L && (q == null || q.isEmpty())) {
					actual.onNext(v);

					if (r != Long.MAX_VALUE) {
						REQUESTED.decrementAndGet(this);
					}

					if (maxConcurrency != Integer.MAX_VALUE) {
						int p = produced + 1;
						if (p == limit) {
							produced = 0;
							s.request(p);
						}
						else {
							produced = p;
						}
					}
				}
				else {
					if (q == null) {
						q = getOrCreateScalarQueue();
					}

					if (!q.offer(v) && failOverflow(v, s)){
						done = true;
						drainLoop();
						return;
					}
				}
				if (WIP.decrementAndGet(this) == 0) {
					if (cancelled) {
						Operators.onDiscard(v, actual.currentContext());
					}
					return;
				}

				drainLoop();
			}
			else {
				Queue<R> q;

				q = getOrCreateScalarQueue();

				if (!q.offer(v) && failOverflow(v, s)) {
					done = true;
				}
				drain(v);
			}
		}

		void tryEmit(FlatMapInner<R> inner, R v) {
			if (wip == 0 && WIP.compareAndSet(this, 0, 1)) {
				long r = requested;

				Queue<R> q = inner.queue;
				if (r != 0 && (q == null || q.isEmpty())) {
					actual.onNext(v);

					if (r != Long.MAX_VALUE) {
						REQUESTED.decrementAndGet(this);
					}

					inner.request(1);
				}
				else {
					if (q == null) {
						q = getOrCreateInnerQueue(inner);
					}

					if (!q.offer(v) && failOverflow(v, inner)){
						inner.done = true;
						drainLoop();
						return;
					}
				}
				if (WIP.decrementAndGet(this) == 0) {
					if (cancelled) {
						Operators.onDiscard(v, actual.currentContext());
					}
					return;
				}

				drainLoop();
			}
			else {
				Queue<R> q = getOrCreateInnerQueue(inner);

				if (!q.offer(v) && failOverflow(v, inner)) {
					inner.done = true;
				}
				drain(v);
			}
		}

		void drain(@Nullable R dataSignal) {
			if (WIP.getAndIncrement(this) != 0) {
				if (dataSignal != null && cancelled) {
					Operators.onDiscard(dataSignal, actual.currentContext());
				}
				return;
			}
			drainLoop();
		}

		void drainLoop() {
			int missed = 1;

			final Subscriber<? super R> a = actual;

			for (; ; ) {

				boolean d = done;

				FlatMapInner<R>[] as = get();

				int n = as.length;

				Queue<R> sq = scalarQueue;

				boolean noSources = isEmpty();

				if (checkTerminated(d, noSources && (sq == null || sq.isEmpty()), a, null)) {
					return;
				}

				boolean again = false;

				long r = requested;
				long e = 0L;
				long replenishMain = 0L;

				if (r != 0L && sq != null) {

					while (e != r) {
						d = done;

						R v = sq.poll();

						boolean empty = v == null;

						if (checkTerminated(d, false, a, v)) {
							return;
						}

						if (empty) {
							break;
						}

						a.onNext(v);

						e++;
					}

					if (e != 0L) {
						replenishMain += e;
						if (r != Long.MAX_VALUE) {
							r = REQUESTED.addAndGet(this, -e);
						}
						e = 0L;
						again = true;
					}
				}
				if (r != 0L && !noSources) {

					int j = lastIndex;
					//Do not need to wrap j since lastIndex is always 0..<n
//					if (j >= n) {
//						j = 0;
//					}

					for (int i = 0; i < n; i++) {
						if (cancelled) {
							Operators.onDiscardQueueWithClear(scalarQueue, actual.currentContext(), null);
							scalarQueue = null;
							s.cancel();
							unsubscribe();
							return;
						}

						FlatMapInner<R> inner = as[j];
						if (inner != null) {
							d = inner.done;
							Queue<R> q = inner.queue;
							if (d && q == null) {
								remove(inner.index);
								again = true;
								replenishMain++;
							}
							else if (q != null) {
								while (e != r) {
									d = inner.done;

									R v;

									try {
										v = q.poll();
									}
									catch (Throwable ex) {
										ex = Operators.onOperatorError(inner, ex,
												actual.currentContext());
										if (!Exceptions.addThrowable(ERROR, this, ex)) {
											Operators.onErrorDropped(ex,
													actual.currentContext());
										}
										v = null;
										d = true;
									}

									boolean empty = v == null;

									if (checkTerminated(d, false, a, v)) {
										return;
									}

									if (d && empty) {
										remove(inner.index);
										again = true;
										replenishMain++;
										break;
									}

									if (empty) {
										break;
									}

									a.onNext(v);

									e++;
								}

								if (e == r) {
									d = inner.done;
									boolean empty = q.isEmpty();
									if (d && empty) {
										remove(inner.index);
										again = true;
										replenishMain++;
									}
								}

								if (e != 0L) {
									if (!inner.done) {
										inner.request(e);
									}
									if (r != Long.MAX_VALUE) {
										r = REQUESTED.addAndGet(this, -e);
										if (r == 0L) {
											break; // 0 .. n - 1
										}
									}
									e = 0L;
								}
							}
						}

						if (r == 0L) {
							break;
						}

						if (++j == n) {
							j = 0;
						}
					}

					lastIndex = j;
				}

				if (r == 0L && !noSources) {
					as = get();
					n = as.length;

					for (int i = 0; i < n; i++) {
						if (cancelled) {
							Operators.onDiscardQueueWithClear(scalarQueue, actual.currentContext(), null);
							scalarQueue = null;
							s.cancel();
							unsubscribe();
							return;
						}

						FlatMapInner<R> inner = as[i];
						if (inner == null) {
							continue;
						}

						d = inner.done;
						Queue<R> q = inner.queue;
						boolean empty = (q == null || q.isEmpty());

						// if we have a non-empty source then quit the cleanup
						if (!empty) {
							break;
						}

						if (d && empty) {
							remove(inner.index);
							again = true;
							replenishMain++;
						}
					}
				}

				if (replenishMain != 0L && !done && !cancelled) {
					s.request(replenishMain);
				}

				if (again) {
					continue;
				}

				missed = WIP.addAndGet(this, -missed);
				if (missed == 0) {
					break;
				}
			}
		}

		boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, @Nullable R value) {
			if (cancelled) {
				Context ctx = actual.currentContext();
				Operators.onDiscard(value, ctx);
				Operators.onDiscardQueueWithClear(scalarQueue, ctx, null);
				scalarQueue = null;
				s.cancel();
				unsubscribe();

				return true;
			}

			if (delayError) {
				if (d && empty) {
					Throwable e = error;
					if (e != null && e != Exceptions.TERMINATED) {
						e = Exceptions.terminate(ERROR, this);
						a.onError(e);
					}
					else {
						a.onComplete();
					}

					return true;
				}
			}
			else {
				if (d) {
					Throwable e = error;
					if (e != null && e != Exceptions.TERMINATED) {
						e = Exceptions.terminate(ERROR, this);
						Context ctx = actual.currentContext();
						Operators.onDiscard(value, ctx);
						Operators.onDiscardQueueWithClear(scalarQueue, ctx, null);
						scalarQueue = null;
						s.cancel();
						unsubscribe();

						a.onError(e);
						return true;
					}
					else if (empty) {
						a.onComplete();
						return true;
					}
				}
			}

			return false;
		}

		void innerError(FlatMapInner<R> inner, Throwable e) {
			e = Operators.onNextInnerError(e, currentContext(), s);
			if(e != null) {
				if (Exceptions.addThrowable(ERROR, this, e)) {
					inner.done = true;
					if (!delayError) {
						done = true;
					}
					drain(null);
				}
				else {
					Operators.onErrorDropped(e, actual.currentContext());
				}
			}
			else {
				drain(null);
			}
		}

		boolean failOverflow(R v, Subscription toCancel){
			Throwable e = Operators.onOperatorError(toCancel,
					Exceptions.failWithOverflow(Exceptions.BACKPRESSURE_ERROR_QUEUE_FULL),
					v, actual.currentContext());

			Operators.onDiscard(v, actual.currentContext());

			if (!Exceptions.addThrowable(ERROR, this, e)) {
				Operators.onErrorDropped(e, actual.currentContext());
				return false;
			}
			return true;
		}

		void innerComplete(FlatMapInner<R> inner) {
			//FIXME temp. reduce the case to empty regular inners
//			if (wip == 0 && WIP.compareAndSet(this, 0, 1)) {
//				Queue<R> queue = inner.queue;
//				if (queue == null || queue.isEmpty()) {
//					remove(inner.index);
//
//					boolean d = done;
//					Queue<R> sq = scalarQueue;
//					boolean noSources = isEmpty();
//
//					if (checkTerminated(d,
//							noSources && (sq == null || sq.isEmpty()),
//							actual)) {
//						return;
//					}
//
//					s.request(1);
//					if (WIP.decrementAndGet(this) != 0) {
//						drainLoop();
//					}
//					return;
//				}
//			}
//			else {
				if (WIP.getAndIncrement(this) != 0) {
					return;
				}
//			}
			drainLoop();
		}

		Queue<R> getOrCreateInnerQueue(FlatMapInner<R> inner) {
			Queue<R> q = inner.queue;
			if (q == null) {
				q = innerQueueSupplier.get();
				inner.queue = q;
			}
			return q;
		}

	}

	static final class FlatMapInner<R>
			implements InnerConsumer<R>, Subscription {

		final FlatMapMain<?, R> parent;

		final int prefetch;

		final int limit;

		volatile Subscription s;
		@SuppressWarnings("rawtypes")
		static final AtomicReferenceFieldUpdater<FlatMapInner, Subscription> S =
				AtomicReferenceFieldUpdater.newUpdater(FlatMapInner.class,
						Subscription.class,
						"s");

		long produced;

		volatile Queue<R> queue;

		volatile boolean done;

		
Represents the optimization mode of this inner subscriber.

/**
		 * Represents the optimization mode of this inner subscriber.
		 */
		int sourceMode;

		int index;

		FlatMapInner(FlatMapMain<?, R> parent, int prefetch) {
			this.parent = parent;
			this.prefetch = prefetch;
//			this.limit = prefetch >> 2;
			this.limit = Operators.unboundedOrLimit(prefetch);
		}

		@Override
		public void onSubscribe(Subscription s) {
			if (Operators.setOnce(S, this, s)) {
				if (s instanceof Fuseable.QueueSubscription) {
					@SuppressWarnings("unchecked") Fuseable.QueueSubscription<R> f =
							(Fuseable.QueueSubscription<R>) s;
					int m = f.requestFusion(Fuseable.ANY | Fuseable.THREAD_BARRIER);
					if (m == Fuseable.SYNC) {
						sourceMode = Fuseable.SYNC;
						queue = f;
						done = true;
						parent.drain(null);
						return;
					}
					if (m == Fuseable.ASYNC) {
						sourceMode = Fuseable.ASYNC;
						queue = f;
					}
					// NONE is just fall-through as the queue will be created on demand
				}
				s.request(Operators.unboundedOrPrefetch(prefetch));
			}
		}

		@Override
		public void onNext(R t) {
			if (sourceMode == Fuseable.ASYNC) {
				parent.drain(t);
			}
			else {
				if (done) {
					Operators.onNextDropped(t, parent.currentContext());
					return;
				}

				if (s == Operators.cancelledSubscription()) {
					Operators.onDiscard(t, parent.currentContext());
					return;
				}

				parent.tryEmit(this, t);
			}
		}

		@Override
		public void onError(Throwable t) {
			done = true;
			parent.innerError(this, t);
		}

		@Override
		public void onComplete() {
			// onComplete is practically idempotent so there is no risk due to subscription-race in async mode
			done = true;
			parent.innerComplete(this);
		}

		@Override
		public void request(long n) {
			if (sourceMode == Fuseable.SYNC) {
				return;
			}
			long p = produced + n;
			if (p >= limit) {
				produced = 0L;
				s.request(p);
			}
			else {
				produced = p;
			}
		}

		@Override
		public Context currentContext() {
			return parent.currentContext();
		}

		@Override
		public void cancel() {
			Operators.terminate(S, this);
			Operators.onDiscardQueueWithClear(queue, parent.currentContext(), null);
		}

		@Override
		@Nullable
		public Object scanUnsafe(Attr key) {
			if (key == Attr.PARENT) return s;
			if (key == Attr.ACTUAL) return parent;
			if (key == Attr.TERMINATED) return done && (queue == null || queue.isEmpty());
			if (key == Attr.CANCELLED) return s == Operators.cancelledSubscription();
			if (key == Attr.BUFFERED) return queue == null ? 0 : queue.size();
			if (key == Attr.PREFETCH) return prefetch;
			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;

			return null;
		}
	}
}

abstract class FlatMapTracker<T> {

	volatile T[] array = empty();

	int[] free = FREE_EMPTY;

	long producerIndex;
	long consumerIndex;

	volatile int size;

	@SuppressWarnings("rawtypes")
	static final AtomicIntegerFieldUpdater<FlatMapTracker> SIZE =
			AtomicIntegerFieldUpdater.newUpdater(FlatMapTracker.class, "size");

	static final int[] FREE_EMPTY = new int[0];

	abstract T[] empty();

	abstract T[] terminated();

	abstract T[] newArray(int size);

	abstract void unsubscribeEntry(T entry);

	abstract void setIndex(T entry, int index);

	final void unsubscribe() {
		T[] a;
		T[] t = terminated();
		synchronized (this) {
			a = array;
			if (a == t) {
				return;
			}
			SIZE.lazySet(this, 0);
			free = null;
			array = t;
		}
		for (T e : a) {
			if (e != null) {
				unsubscribeEntry(e);
			}
		}
	}

	final T[] get() {
		return array;
	}

	final boolean add(T entry) {
		T[] a = array;
		if (a == terminated()) {
			return false;
		}
		synchronized (this) {
			a = array;
			if (a == terminated()) {
				return false;
			}

			int idx = pollFree();
			if (idx < 0) {
				int n = a.length;
				T[] b = n != 0 ? newArray(n << 1) : newArray(4);
				System.arraycopy(a, 0, b, 0, n);

				array = b;
				a = b;

				int m = b.length;
				int[] u = new int[m];
				for (int i = n + 1; i < m; i++) {
					u[i] = i;
				}
				free = u;
				consumerIndex = n + 1;
				producerIndex = m;

				idx = n;
			}
			setIndex(entry, idx);
			SIZE.lazySet(this, size); // make sure entry is released
			a[idx] = entry;
			SIZE.lazySet(this, size + 1);
		}
		return true;
	}

	final void remove(int index) {
		synchronized (this) {
			T[] a = array;
			if (a != terminated()) {
				a[index] = null;
				offerFree(index);
				SIZE.lazySet(this, size - 1);
			}
		}
	}

	int pollFree() {
		int[] a = free;
		int m = a.length - 1;
		long ci = consumerIndex;
		if (producerIndex == ci) {
			return -1;
		}
		int offset = (int) ci & m;
		consumerIndex = ci + 1;
		return a[offset];
	}

	void offerFree(int index) {
		int[] a = free;
		int m = a.length - 1;
		long pi = producerIndex;
		int offset = (int) pi & m;
		a[offset] = index;
		producerIndex = pi + 1;
	}

	final boolean isEmpty() {
		return size == 0;
	}
}