import java.util.concurrent.Flow; //导入依赖的package包/类
@Test
public void cancelRequests() {
    SubscriptionArbiter sa = new SubscriptionArbiter();

    sa.arbiterReplace(new Flow.Subscription() {

        @Override
        public void request(long n) {
        }

        @Override
        public void cancel() {
            sa.request(1);
        }
    });

    sa.cancel();
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Test
public void mapperThrows() {
    DirectProcessor<Integer> bp = new DirectProcessor<>();

    TestConsumer<Object> tc = bp
            .switchFlatMap((CheckedFunction<Object, Flow.Publisher<Object>>) v -> {
                throw new IOException();
            }, 2)
            .test();

    bp.onNext(1);

    tc.assertFailure(IOException.class);

    assertFalse(bp.hasSubscribers());
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public void onSubscribe(Flow.Subscription s) {
    this.s = s;

    try {
        parent.onSubscribe.accept(s);
    } catch (Throwable ex) {
        FolyamPlugins.handleFatal(ex);
        s.cancel();
        actual.onSubscribe(EmptySubscription.INSTANCE);
        onError(ex);
        return;
    }

    actual.onSubscribe(this);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
protected void subscribeActual(FolyamSubscriber<? super T> s) {
    Flow.Publisher<? extends T>[] array = new Flow.Publisher[8];
    int n = 0;
    try {
        for (Flow.Publisher<? extends T> p : sourcesIterable) {
            if (n == array.length) {
                array = Arrays.copyOf(array, n << 1);
            }
            array[n++] = Objects.requireNonNull(p, "a source is null");
        }
    } catch (Throwable ex) {
        FolyamPlugins.handleFatal(ex);
        EmptySubscription.error(s, ex);
        return;
    }

    FolyamOrderedMergeArray.subscribe(s, array, n, comparator, prefetch, delayErrors);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public Flow.Publisher<Integer> createFlowPublisher(final long elements) {
    final LastProcessor<Integer> pp = new LastProcessor<>();

    SchedulerServices.io().schedule(() -> {
        long start = System.currentTimeMillis();
        while (!pp.hasSubscribers()) {
            try {
                Thread.sleep(1);
            } catch (InterruptedException ex) {
                return;
            }

            if (System.currentTimeMillis() - start > 200) {
                return;
            }
        }

        for (int i = 0; i < elements; i++) {
            pp.onNext(i);
        }
        pp.onComplete();
    });
    return pp;
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public void onError(Throwable throwable) {
    if (once) {
        actual.onError(throwable);
        return;
    }
    once = true;
    Flow.Publisher<? extends T> p;

    try {
        p = Objects.requireNonNull(handler.apply(throwable), "The handler returned a null Flow.Publisher");
    } catch (Throwable ex) {
        actual.onError(new CompositeThrowable(throwable, ex));
        return;
    }

    long c = produced;
    if (c != 0) {
        arbiterProduced(c);
    }
    p.subscribe(this);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
public final void onSubscribe(Flow.Subscription subscription) {
    upstream = subscription;
    arbiterReplace(subscription);
    if (firstTimeout != null) {
        AtomicDisposable d = new AtomicDisposable();
        task = d;

        onStart();
        if (d.get() == null) {
            TimeoutInnerSubscriber inner = new TimeoutInnerSubscriber(this, 0);
            if (d.compareAndSet(null, inner)) {
                firstTimeout.subscribe(inner);
            }
        }
    } else {
        onStart();
    }
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
public final void onNext(T item) {
    AutoDisposable d = task;
    if (d != null) {
        d.close();
    }

    long idx = (long)INDEX.getAcquire(this);
    if (idx != Long.MIN_VALUE && INDEX.compareAndSet(this, idx, idx + 1)) {
        next(item);

        Flow.Publisher<?> p;

        try {
            p = Objects.requireNonNull(itemTimeoutSelector.apply(item), "The itemTimeoutSelector returned a null Flow.Publisher");
        } catch (Throwable ex) {
            onError(ex);
            return;
        }
        // replace
        TimeoutInnerSubscriber inner = new TimeoutInnerSubscriber(this, idx + 1);
        if (DisposableHelper.replace(this, TASK, inner)) {
            p.subscribe(inner);
        }
    }
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
void subscribe(Flow.Publisher<? extends T>[] sources, int n) {
    for (int i = 0; i < n; i++) {
        if (cancelled) {
            return;
        }
        sources[i].subscribe(subscribers[i]);
    }
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
public void onSubscribe(Flow.Subscription subscription) {
    //count = bufferSize - bufferSize / 2;// 当消费一半的时候重新请求
    (this.subscription = subscription).request(bufferSize);
    System.out.println("开始onSubscribe订阅");
    try {
        Thread.sleep(100);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
public FolyamConcatMapEager(Folyam<T> source, CheckedFunction<? super T, ? extends Flow.Publisher<? extends R>> mapper, int maxConcurrency, int prefetch, boolean delayError) {
    this.source = source;
    this.mapper = mapper;
    this.maxConcurrency = maxConcurrency;
    this.prefetch = prefetch;
    this.delayError = delayError;
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
protected DoOnSignalSubscriber(FolyamSubscriber<? super T> actual,
                               CheckedConsumer<? super Flow.Subscription> onSubscribe,
                               CheckedConsumer<? super T> onNext,
                               CheckedConsumer<? super T> onAfterNext,
                               CheckedConsumer<? super Throwable> onError,
                               CheckedRunnable onComplete,
                               CheckedConsumer<? super Long> onRequest,
                               CheckedRunnable onCancel) {
    super(onSubscribe, onNext, onAfterNext, onError, onComplete, onRequest, onCancel);
    this.actual = actual;
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void onSubscribe(Flow.Subscription s) {
    this.s = s;
    if (s instanceof FusedSubscription) {
        qs = (FusedSubscription<T>)s;
    }

    calls.onSubscribeCount++;
    startTime = System.nanoTime();

    actual.onSubscribe(this);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
/**
 * Pushes subscriber to subscribers list.
 * @param subscriber
 */
@Override public void subscribe(Flow.Subscriber<? super byte[]> subscriber) {
  if (!started) {
    startReading();
    started = true;
  }

  NoBackpressureSubscription subscription = new NoBackpressureSubscription(this);

  this.subscribers.put(subscriber, subscription);

  subscriber.onSubscribe(subscription);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public void onSubscribe(Flow.Subscription s) {
    this.s = s;

    actual.onSubscribe(this);

    s.request(prefetch);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public final void onSubscribe(Flow.Subscription subscription) {
    upstream = subscription;
    if (subscription instanceof FusedSubscription) {
        FusedSubscription<T> fs = (FusedSubscription<T>) subscription;
        int m = fs.requestFusion(ANY);
        if (m == SYNC) {
            sourceFused = m;
            queue = fs;
            DONE.setRelease(this, true);
            onStart();
            return;
        }
        if (m == ASYNC) {
            sourceFused = m;
            queue = fs;
            onStart();
            fs.request(prefetch);
            return;
        }
    }

    int pf = prefetch;
    if (pf == 1) {
        queue = new SpscOneQueue<>();
    } else {
        queue = new SpscArrayQueue<>(pf);
    }

    onStart();

    subscription.request(pf);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
protected DoOnSignalConditionalSubscriber(ConditionalSubscriber<? super T> actual,
                               CheckedConsumer<? super Flow.Subscription> onSubscribe,
                               CheckedConsumer<? super T> onNext,
                               CheckedConsumer<? super T> onAfterNext,
                               CheckedConsumer<? super Throwable> onError,
                               CheckedRunnable onComplete,
                               CheckedConsumer<? super Long> onRequest,
                               CheckedRunnable onCancel) {
    super(onSubscribe, onNext, onAfterNext, onError, onComplete, onRequest, onCancel);
    this.actual = actual;
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
/**
 * Converts a Flow Publisher into a Reactive Streams Publisher.
 * @param <T> the element type
 * @param flowPublisher the source Flow Publisher to convert
 * @return the equivalent Reactive Streams Publisher
 */
@SuppressWarnings("unchecked")
public static <T> org.reactivestreams.Publisher<T> toReactiveStreams(
        Flow.Publisher<? extends T> flowPublisher) {
    if (flowPublisher == null) {
        throw new NullPointerException("flowPublisher");
    }
    if (flowPublisher instanceof org.reactivestreams.Publisher) {
        return (org.reactivestreams.Publisher<T>)flowPublisher;
    }
    if (flowPublisher instanceof FlowPublisherFromReactive) {
        return (org.reactivestreams.Publisher<T>)(((FlowPublisherFromReactive<T>)flowPublisher).reactiveStreams);
    }
    return new ReactivePublisherFromFlow<T>(flowPublisher);
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
public static void onComplete(Flow.Subscriber<?> actual, Object target, VarHandle WIP, VarHandle ERRORS) {
    if ((int) WIP.getAndAdd(target, 1) == 0) {
        Throwable ex = ExceptionHelper.terminate(target, ERRORS);
        if (ex == null) {
            actual.onComplete();
        } else {
            actual.onError(ex);
        }
    }
}
 

import java.util.concurrent.Flow; //导入依赖的package包/类
@Override
public Flow.Publisher<Long> createFlowPublisher(long elements) {
    return
        Folyam.concat(Folyam.fromArray(
                Folyam.fromIterable(iterate(elements / 2)),
                Folyam.fromIterable(iterate(elements - elements / 2))
            )
        )
    ;
}