<commit_before>package com.github.davidmoten.rx.internal.operators;

import com.github.davidmoten.util.Preconditions;

import rx.Notification;
import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observable.Transformer;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.Func3;

public final class TransformerStateMachine<State, In, Out> implements Transformer<In, Out> {

    private final Func0<? extends State> initialState;
    private final Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition;
    private final Func2<? super State, ? super Subscriber<Out>, Boolean> completion;
    private final Transformer<Notification<Out>, Notification<Out>> backpressureStrategy;

    private TransformerStateMachine(Func0<? extends State> initialState,
            Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {
        Preconditions.checkNotNull(initialState);
        Preconditions.checkNotNull(transition);
        Preconditions.checkNotNull(completion);
        Preconditions.checkNotNull(backpressureStrategy);
        this.initialState = initialState;
        this.transition = transition;
        this.completion = completion;
        this.backpressureStrategy = backpressureStrategy;
    }

    public static <State, In, Out> Transformer<In, Out> create(Func0<? extends State> initialState,
            Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {
        return new TransformerStateMachine<State, In, Out>(initialState, transition, completion,
                backpressureStrategy);
    }

    @Override
    public Observable<Out> call(final Observable<In> source) {
        // use defer so we can have a single state reference for each
        // subscription
        return Observable.defer(new Func0<Observable<Out>>() {
            @Override
            public Observable<Out> call() {
                Mutable<State> state = new Mutable<State>(initialState.call());
                return source.materialize()
                        // do state transitions and emit notifications
                        // use flatMap to emit notification values
                        .flatMap(execute(transition, completion, state, backpressureStrategy))
                        // filter out unsubscribed error notification that is
                        // generated by unsubscribing in the transition
                        .materialize()
                        // don't pass through an unsubscribed exception (its
                        // just flow control not a real exception)
                        .takeWhile(NOT_UNSUBSCRIBED)
                        // flatten notifications of notifications to
                        // notifications
                        .dematerialize()
                        // flatten notifications to a stream which will enable
                        // early termination from the state machine if desired
                        .dematerialize();
            }
        });
    }

    private static final Func1<Notification<?>, Boolean> NOT_UNSUBSCRIBED = new Func1<Notification<?>, Boolean>() {

        @Override
        public Boolean call(Notification<?> t) {
            return !t.isOnError() || t.getThrowable() != UnsubscribedExceptionHolder.INSTANCE;
        }

    };

    private static final class Mutable<T> {
        // mutable
        T value;

        Mutable(T value) {
            this.value = value;
        }
    }

    private static final class UnsubscribedExceptionHolder {
        static final UnsubscribedException INSTANCE = new UnsubscribedException();
    }

    private static class UnsubscribedException extends RuntimeException {
        private static final long serialVersionUID = 7177818068143224232L;
    }

    private static <State, Out, In> Func1<Notification<In>, Observable<Notification<Out>>> execute(
            final Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            final Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            final Mutable<State> state,
            final Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {

        return new Func1<Notification<In>, Observable<Notification<Out>>>() {

            @Override
            public Observable<Notification<Out>> call(final Notification<In> in) {

                return Observable.create(new OnSubscribe<Notification<Out>>() {

                    @Override
                    public void call(Subscriber<? super Notification<Out>> subscriber) {
                        Subscriber<Out> w = wrap(subscriber);
                        if (in.hasValue()) {
                            state.value = transition.call(state.value, in.getValue(), w);
                            if (!subscriber.isUnsubscribed())
                                subscriber.onCompleted();
                            else {
                                subscriber.onError(UnsubscribedExceptionHolder.INSTANCE);
                            }
                        } else if (in.isOnCompleted()) {
                            if (completion.call(state.value, w)) {
                                if (!subscriber.isUnsubscribed())
                                    w.onCompleted();
                            }
                        } else if (!subscriber.isUnsubscribed()) {
                            w.onError(in.getThrowable());
                        }
                    }
                }).compose(backpressureStrategy);
            }

        };
    }

    private static <Out> NotificationSubscriber<Out> wrap(
            Subscriber<? super Notification<Out>> sub) {
        return new NotificationSubscriber<Out>(sub);
    }

    private static final class NotificationSubscriber<Out> extends Subscriber<Out> {

        private final Subscriber<? super Notification<Out>> sub;

        NotificationSubscriber(Subscriber<? super Notification<Out>> sub) {
            this.sub = sub;
            add(sub);
        }<commit_msg>add comments and change method order<commit_after>package com.github.davidmoten.rx.internal.operators;

import com.github.davidmoten.util.Preconditions;

import rx.Notification;
import rx.Observable;
import rx.Observable.OnSubscribe;
import rx.Observable.Transformer;
import rx.Subscriber;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.functions.Func3;

public final class TransformerStateMachine<State, In, Out> implements Transformer<In, Out> {

    private final Func0<? extends State> initialState;
    private final Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition;
    private final Func2<? super State, ? super Subscriber<Out>, Boolean> completion;
    private final Transformer<Notification<Out>, Notification<Out>> backpressureStrategy;

    private TransformerStateMachine(Func0<? extends State> initialState,
            Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {
        Preconditions.checkNotNull(initialState);
        Preconditions.checkNotNull(transition);
        Preconditions.checkNotNull(completion);
        Preconditions.checkNotNull(backpressureStrategy);
        this.initialState = initialState;
        this.transition = transition;
        this.completion = completion;
        this.backpressureStrategy = backpressureStrategy;
    }

    public static <State, In, Out> Transformer<In, Out> create(Func0<? extends State> initialState,
            Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {
        return new TransformerStateMachine<State, In, Out>(initialState, transition, completion,
                backpressureStrategy);
    }

    @Override
    public Observable<Out> call(final Observable<In> source) {
        // use defer so we can have a single state reference for each
        // subscription
        return Observable.defer(new Func0<Observable<Out>>() {
            @Override
            public Observable<Out> call() {
                Mutable<State> state = new Mutable<State>(initialState.call());
                return source.materialize()
                        // do state transitions and emit notifications
                        // use flatMap to emit notification values
                        .flatMap(execute(transition, completion, state, backpressureStrategy))
                        // filter out unsubscribed error notification that is
                        // generated by unsubscribing in the transition
                        .materialize()
                        // don't pass through an unsubscribed exception (its
                        // just flow control not a real exception)
                        .takeWhile(NOT_UNSUBSCRIBED)
                        // flatten notifications of notifications to
                        // notifications
                        .dematerialize()
                        // flatten notifications to a stream which will enable
                        // early termination from the state machine if desired
                        .dematerialize();
            }
        });
    }

    private static <State, Out, In> Func1<Notification<In>, Observable<Notification<Out>>> execute(
            final Func3<? super State, ? super In, ? super Subscriber<Out>, ? extends State> transition,
            final Func2<? super State, ? super Subscriber<Out>, Boolean> completion,
            final Mutable<State> state,
            final Transformer<Notification<Out>, Notification<Out>> backpressureStrategy) {

        return new Func1<Notification<In>, Observable<Notification<Out>>>() {

            @Override
            public Observable<Notification<Out>> call(final Notification<In> in) {

                return Observable.create(new OnSubscribe<Notification<Out>>() {

                    @Override
                    public void call(Subscriber<? super Notification<Out>> subscriber) {
                        Subscriber<Out> w = wrap(subscriber);
                        if (in.hasValue()) {
                            state.value = transition.call(state.value, in.getValue(), w);
                            if (!subscriber.isUnsubscribed())
                                subscriber.onCompleted();
                            else {
                                subscriber.onError(UnsubscribedExceptionHolder.INSTANCE);
                            }
                        } else if (in.isOnCompleted()) {
                            if (completion.call(state.value, w) && !subscriber.isUnsubscribed()) {
                                w.onCompleted();
                            }
                        } else if (!subscriber.isUnsubscribed()) {
                            w.onError(in.getThrowable());
                        }
                    }
                })
                        // because the observable we just created does not
                        // support backpressure we need to apply a backpressure
                        // handling operator. This operator is supplied by the
                        // user.
                        .compose(backpressureStrategy);
            }

        };
    }

    private static final Func1<Notification<?>, Boolean> NOT_UNSUBSCRIBED = new Func1<Notification<?>, Boolean>() {

        @Override
        public Boolean call(Notification<?> t) {
            return !t.isOnError() || t.getThrowable() != UnsubscribedExceptionHolder.INSTANCE;
        }

    };

    private static final class Mutable<T> {
        T value;

        Mutable(T value) {
            this.value = value;
        }
    }

    private static final class UnsubscribedExceptionHolder {
        static final UnsubscribedException INSTANCE = new UnsubscribedException();
    }

    private static class UnsubscribedException extends RuntimeException {
        private static final long serialVersionUID = 7177818068143224232L;
    }

    private static <Out> NotificationSubscriber<Out> wrap(
            Subscriber<? super Notification<Out>> sub) {
        return new NotificationSubscriber<Out>(sub);
    }

    private static final class NotificationSubscriber<Out> extends Subscriber<Out> {

        private final Subscriber<? super Notification<Out>> sub;

        NotificationSubscriber(Subscriber<? super Notification<Out>> sub) {
            this.sub = sub;
            add(sub);
        }
{'fraction_non_alphanumeric': 0.0872189527689754, 'fraction_numerical': 0.005885015844273427, 'mean_word_length': 2.027638190954774, 'pattern_counts': {'":': 0, '<': 159, '<?xml version=': 0, '>': 159, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1468360', 'n_tokens_mistral': 3201, 'n_tokens_neox': 2984, 'n_words': 842}