/*
 * 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
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package reactor.core.publisher;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Function;

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

A Flux that emits items from upstream and recursively expand them into inner sequences that are also replayed. The type of recursion is driven by the breadthFirst parameter.
Author:David Karnok, Simon Baslé
Type parameters:
  • <T> –
/** * A {@link Flux} that emits items from upstream and recursively expand them into * inner sequences that are also replayed. The type of recursion is driven by the * {@code breadthFirst} parameter. * * @param <T> * * @author David Karnok * @author Simon Baslé */
//adapted from RxJava2Extensions: https://github.com/akarnokd/RxJava2Extensions/blob/master/src/main/java/hu/akarnokd/rxjava2/operators/FlowableExpand.java final class FluxExpand<T> extends FluxOperator<T, T> { final boolean breadthFirst; final Function<? super T, ? extends Publisher<? extends T>> expander; final int capacityHint; FluxExpand(Flux<T> source, Function<? super T, ? extends Publisher<? extends T>> expander, boolean breadthFirst, int capacityHint) { super(source); this.expander = expander; this.breadthFirst = breadthFirst; this.capacityHint = capacityHint; } @Override public void subscribe(CoreSubscriber<? super T> s) { if (breadthFirst) { ExpandBreathSubscriber<T> parent = new ExpandBreathSubscriber<>(s, expander, capacityHint); parent.queue.offer(source); s.onSubscribe(parent); parent.drainQueue(); } else { ExpandDepthSubscription<T> parent = new ExpandDepthSubscription<>(s, expander, capacityHint); parent.source = source; s.onSubscribe(parent); } } static final class ExpandBreathSubscriber<T> extends Operators.MultiSubscriptionSubscriber<T, T> { final Function<? super T, ? extends Publisher<? extends T>> expander; final Queue<Publisher<? extends T>> queue; volatile boolean active; volatile int wip; static final AtomicIntegerFieldUpdater<ExpandBreathSubscriber> WIP = AtomicIntegerFieldUpdater.newUpdater(ExpandBreathSubscriber.class, "wip"); long produced; ExpandBreathSubscriber(CoreSubscriber<? super T> actual, Function<? super T, ? extends Publisher<? extends T>> expander, int capacityHint) { super(actual); this.expander = expander; this.queue = Queues.<Publisher<? extends T>>unbounded(capacityHint).get(); } @Override public void onSubscribe(Subscription s) { set(s); } @Override public void onNext(T t) { produced++; actual.onNext(t); Publisher<? extends T> p; try { p = Objects.requireNonNull(expander.apply(t), "The expander returned a null Publisher"); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); super.cancel(); actual.onError(ex); drainQueue(); return; } queue.offer(p); } @Override public void onError(Throwable t) { set(Operators.cancelledSubscription()); super.cancel(); actual.onError(t); drainQueue(); } @Override public void onComplete() { active = false; drainQueue(); } @Override public void cancel() { super.cancel(); drainQueue(); } void drainQueue() { if (WIP.getAndIncrement(this) == 0) { do { Queue<Publisher<? extends T>> q = queue; if (isCancelled()) { q.clear(); } else { if (!active) { if (q.isEmpty()) { set(Operators.cancelledSubscription()); super.cancel(); actual.onComplete(); } else { Publisher<? extends T> p = q.poll(); long c = produced; if (c != 0L) { produced = 0L; produced(c); } active = true; p.subscribe(this); } } } } while (WIP.decrementAndGet(this) != 0); } } @Nullable @Override public Object scanUnsafe(Attr key) { if (key == Attr.BUFFERED) return queue != null ? queue.size() : 0; return super.scanUnsafe(key); } } static final class ExpandDepthSubscription<T> implements InnerProducer<T> { final CoreSubscriber<? super T> actual; final Function<? super T, ? extends Publisher<? extends T>> expander; volatile Throwable error; static final AtomicReferenceFieldUpdater<ExpandDepthSubscription, Throwable> ERROR = AtomicReferenceFieldUpdater.newUpdater(ExpandDepthSubscription.class, Throwable.class, "error"); volatile int active; static final AtomicIntegerFieldUpdater<ExpandDepthSubscription> ACTIVE = AtomicIntegerFieldUpdater.newUpdater(ExpandDepthSubscription.class, "active"); volatile long requested; static final AtomicLongFieldUpdater<ExpandDepthSubscription> REQUESTED = AtomicLongFieldUpdater.newUpdater(ExpandDepthSubscription.class, "requested"); volatile Object current; static final AtomicReferenceFieldUpdater<ExpandDepthSubscription, Object> CURRENT = AtomicReferenceFieldUpdater.newUpdater(ExpandDepthSubscription.class, Object.class, "current"); volatile int wip; static final AtomicIntegerFieldUpdater<ExpandDepthSubscription> WIP = AtomicIntegerFieldUpdater.newUpdater(ExpandDepthSubscription.class, "wip"); Deque<ExpandDepthSubscriber<T>> subscriptionStack; volatile boolean cancelled; Publisher<? extends T> source; long consumed; ExpandDepthSubscription(CoreSubscriber<? super T> actual, Function<? super T, ? extends Publisher<? extends T>> expander, int capacityHint) { this.actual = actual; this.expander = expander; this.subscriptionStack = new ArrayDeque<>(capacityHint); } @Override public CoreSubscriber<? super T> actual() { return actual; } @Override public void request(long n) { if (Operators.validate(n)) { Operators.addCap(REQUESTED, this, n); drainQueue(); } } @SuppressWarnings("unchecked") @Override public void cancel() { if (!cancelled) { cancelled = true; Deque<ExpandDepthSubscriber<T>> q; synchronized (this) { q = subscriptionStack; subscriptionStack = null; } if (q != null) { while (!q.isEmpty()) { q.poll().dispose(); } } Object o = CURRENT.getAndSet(this, this); if (o != this && o != null) { ((ExpandDepthSubscriber) o).dispose(); } } } @Nullable ExpandDepthSubscriber<T> pop() { synchronized (this) { Deque<ExpandDepthSubscriber<T>> q = subscriptionStack; return q != null ? q.pollFirst() : null; } } boolean push(ExpandDepthSubscriber<T> subscriber) { synchronized (this) { Deque<ExpandDepthSubscriber<T>> q = subscriptionStack; if (q != null) { q.offerFirst(subscriber); return true; } return false; } } boolean setCurrent(ExpandDepthSubscriber<T> inner) { for (;;) { Object o = CURRENT.get(this); if (o == this) { inner.dispose(); return false; } if (CURRENT.compareAndSet(this, o, inner)) { return true; } } } void drainQueue() { if (WIP.getAndIncrement(this) != 0) { return; } int missed = 1; Subscriber<? super T> a = actual; long e = consumed; for (;;) { Object o = current; if (cancelled || o == this) { source = null; return; } @SuppressWarnings("unchecked") ExpandDepthSubscriber<T> curr = (ExpandDepthSubscriber<T>)o; Publisher<? extends T> p = source; if (curr == null && p != null) { source = null; ACTIVE.getAndIncrement(this); ExpandDepthSubscriber<T> eds = new ExpandDepthSubscriber<>(this); if (setCurrent(eds)) { p.subscribe(eds); } else { return; } } else if (curr == null) { return; } else { boolean currentDone = curr.done; T v = curr.value; boolean newSource = false; if (v != null && e != requested) { curr.value = null; a.onNext(v); e++; try { p = Objects.requireNonNull(expander.apply(v), "The expander returned a null Publisher"); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); p = null; curr.dispose(); curr.done = true; currentDone = true; v = null; Exceptions.addThrowable(ERROR, this, ex); } if (p != null) { if (push(curr)) { ACTIVE.getAndIncrement(this); curr = new ExpandDepthSubscriber<>(this); if (setCurrent(curr)) { p.subscribe(curr); newSource = true; } else { return; } } } } if (!newSource) { if (currentDone && v == null) { if (ACTIVE.decrementAndGet(this) == 0) { Throwable ex = Exceptions.terminate(ERROR, this); if (ex != null) { a.onError(ex); } else { a.onComplete(); } return; } curr = pop(); if (curr != null && setCurrent(curr)) { curr.requestOne(); continue; } else { return; } } } } int w = wip; if (missed == w) { consumed = e; missed = WIP.addAndGet(this, -missed); if (missed == 0) { break; } } else { missed = w; } } } void innerNext() { drainQueue(); } void innerError(ExpandDepthSubscriber inner, Throwable t) { Exceptions.addThrowable(ERROR, this, t); inner.done = true; drainQueue(); } void innerComplete(ExpandDepthSubscriber inner) { inner.done = true; drainQueue(); } @Nullable @Override public Object scanUnsafe(Attr key) { if (key == Attr.CANCELLED) return cancelled; if (key == Attr.REQUESTED_FROM_DOWNSTREAM) return this.requested; if (key == Attr.ERROR) return this.error; return InnerProducer.super.scanUnsafe(key); } } static final class ExpandDepthSubscriber<T> implements InnerConsumer<T> { ExpandDepthSubscription<T> parent; volatile boolean done; volatile T value; volatile Subscription s; static final AtomicReferenceFieldUpdater<ExpandDepthSubscriber, Subscription> S = AtomicReferenceFieldUpdater.newUpdater(ExpandDepthSubscriber.class, Subscription.class, "s"); ExpandDepthSubscriber(ExpandDepthSubscription<T> parent) { this.parent = parent; } @Override public void onSubscribe(Subscription s) { if (Operators.setOnce(S, this, s)) { s.request(1); } } @Override public void onNext(T t) { if (s != Operators.cancelledSubscription()) { value = t; parent.innerNext(); } } @Override public void onError(Throwable t) { if (s != Operators.cancelledSubscription()) { parent.innerError(this, t); } } @Override public void onComplete() { if (s != Operators.cancelledSubscription()) { parent.innerComplete(this); } } void requestOne() { s.request(1); } void dispose() { Operators.terminate(S, this); } @Nullable @Override public Object scanUnsafe(Attr key) { if (key == Attr.PARENT) return s; if (key == Attr.ACTUAL) return parent.actual; if (key == Attr.TERMINATED) return this.done; return null; } @Override public Context currentContext() { return parent.actual().currentContext(); } } }