package reactivestreams.commons.publisher;

import java.util.function.Consumer;
import java.util.function.LongConsumer;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactivestreams.commons.flow.Fuseable;
import reactivestreams.commons.flow.Fuseable.ConditionalSubscriber;
import reactivestreams.commons.flow.Producer;
import reactivestreams.commons.flow.Receiver;
import reactivestreams.commons.publisher.PublisherPeekFuseable.PublisherPeekConditionalSubscriber;
import reactivestreams.commons.publisher.PublisherPeekFuseable.PublisherPeekFuseableSubscriber;
import reactivestreams.commons.util.EmptySubscription;
import reactivestreams.commons.util.ExceptionHelper;

/**
 * Peek into the lifecycle events and signals of a sequence.
 * <p>
 * <p>
 * The callbacks are all optional.
 * <p>
 * <p>
 * Crashes by the lambdas are ignored.
 *
 * @param <T> the value type
 */
public final class PublisherPeek<T> extends PublisherSource<T, T>
implements PublisherPeekHelper<T> {

    final Consumer<? super Subscription> onSubscribeCall;

    final Consumer<? super T> onNextCall;

    final Consumer<? super Throwable> onErrorCall;

    final Runnable onCompleteCall;

    final Runnable onAfterTerminateCall;

    final LongConsumer onRequestCall;

    final Runnable onCancelCall;

    public PublisherPeek(Publisher<? extends T> source, Consumer<? super Subscription> onSubscribeCall,
                         Consumer<? super T> onNextCall, Consumer<? super Throwable> onErrorCall, Runnable
                           onCompleteCall,
                         Runnable onAfterTerminateCall, LongConsumer onRequestCall, Runnable onCancelCall) {
        super(source);
        this.onSubscribeCall = onSubscribeCall;
        this.onNextCall = onNextCall;
        this.onErrorCall = onErrorCall;
        this.onCompleteCall = onCompleteCall;
        this.onAfterTerminateCall = onAfterTerminateCall;
        this.onRequestCall = onRequestCall;
        this.onCancelCall = onCancelCall;
    }

    @Override
    @SuppressWarnings("unchecked")
    public void subscribe(Subscriber<? super T> s) {
        if (source instanceof Fuseable) {
            source.subscribe(new PublisherPeekFuseableSubscriber<>(s, this));
            return;
        }
        if (s instanceof ConditionalSubscriber) {
            source.subscribe(new PublisherPeekConditionalSubscriber<>((ConditionalSubscriber<? super T>)s, this));
            return;
        }
        source.subscribe(new PublisherPeekSubscriber<>(s, this));
    }
    
    static final class PublisherPeekSubscriber<T> implements Subscriber<T>, Subscription, Receiver, Producer {

        final Subscriber<? super T> actual;

        final PublisherPeekHelper<T> parent;

        Subscription s;

        public PublisherPeekSubscriber(Subscriber<? super T> actual, PublisherPeekHelper<T> parent) {
            this.actual = actual;
            this.parent = parent;
        }

        @Override
        public void request(long n) {
            if(parent.onRequestCall() != null) {
                try {
                    parent.onRequestCall().accept(n);
                }
                catch (Throwable e) {
                    cancel();
                    onError(ExceptionHelper.unwrap(e));
                    return;
                }
            }
            s.request(n);
        }

        @Override
        public void cancel() {
            if(parent.onCancelCall() != null) {
                try {
                    parent.onCancelCall().run();
                }
                catch (Throwable e) {
                    ExceptionHelper.throwIfFatal(e);
                    s.cancel();
                    onError(ExceptionHelper.unwrap(e));
                    return;
                }
            }
            s.cancel();
        }

        @Override
        public void onSubscribe(Subscription s) {
            if(parent.onSubscribeCall() != null) {
                try {
                    parent.onSubscribeCall().accept(s);
                }
                catch (Throwable e) {
                    s.cancel();
                    actual.onSubscribe(EmptySubscription.INSTANCE);
                    onError(e);
                    return;
                }
            }
            this.s = s;
            actual.onSubscribe(this);
        }

        @Override
        public void onNext(T t) {
            if(parent.onNextCall() != null) {
                try {
                    parent.onNextCall().accept(t);
                }
                catch (Throwable e) {
                    cancel();
                    ExceptionHelper.throwIfFatal(e);
                    onError(ExceptionHelper.unwrap(e));
                    return;
                }
            }
            actual.onNext(t);
        }

        @Override
        public void onError(Throwable t) {
            if(parent.onErrorCall() != null) {
                ExceptionHelper.throwIfFatal(t);
                parent.onErrorCall().accept(t);
            }

            actual.onError(t);

            if(parent.onAfterTerminateCall() != null) {
                try {
                    parent.onAfterTerminateCall().run();
                }
                catch (Throwable e) {
                    ExceptionHelper.throwIfFatal(e);
                    Throwable _e = ExceptionHelper.unwrap(e);
                    e.addSuppressed(ExceptionHelper.unwrap(t));
                    if(parent.onErrorCall() != null) {
                        parent.onErrorCall().accept(_e);
                    }
                    actual.onError(_e);
                }
            }
        }

        @Override
        public void onComplete() {
            if(parent.onCompleteCall() != null) {
                try {
                    parent.onCompleteCall().run();
                }
                catch (Throwable e) {
                    ExceptionHelper.throwIfFatal(e);
                    onError(ExceptionHelper.unwrap(e));
                    return;
                }
            }

            actual.onComplete();

            if(parent.onAfterTerminateCall() != null) {
                try {
                    parent.onAfterTerminateCall().run();
                }
                catch (Throwable e) {
                    ExceptionHelper.throwIfFatal(e);
                    Throwable _e = ExceptionHelper.unwrap(e);
                    if(parent.onErrorCall() != null) {
                        parent.onErrorCall().accept(_e);
                    }
                    actual.onError(_e);
                }
            }
        }

        @Override
        public Object downstream() {
            return actual;
        }

        @Override
        public Object upstream() {
            return s;
        }
    }

    @Override
    public Consumer<? super Subscription> onSubscribeCall() {
        return onSubscribeCall;
    }



    @Override
    public Consumer<? super T> onNextCall() {
        return onNextCall;
    }



    @Override
    public Consumer<? super Throwable> onErrorCall() {
        return onErrorCall;
    }



    @Override
    public Runnable onCompleteCall() {
        return onCompleteCall;
    }



    @Override
    public Runnable onAfterTerminateCall() {
        return onAfterTerminateCall;
    }



    @Override
    public LongConsumer onRequestCall() {
        return onRequestCall;
    }



    @Override
    public Runnable onCancelCall() {
        return onCancelCall;
    }
}

@@ -2080,43 +2080,8 @@
 ide%0A
-    @SuppressWarnings(%22unchecked%22)%0A
     
@@ -2345,42 +2345,113 @@
     
-source.subscribe(new PublisherPeek
+@SuppressWarnings(%22unchecked%22) // javac, give reason to suppress because inference anomalies%0A            
 Cond
@@ -2468,18 +2468,24 @@
 scriber%3C
-%3E(
+T%3E s2 = 
 (Conditi
@@ -2503,20 +2503,86 @@
 ber%3C
-? super T%3E)s
+T%3E)s;%0A            source.subscribe(new PublisherPeekConditionalSubscriber%3C%3E(s2
 , th

Fix suppress