package reactfx;

import javafx.beans.InvalidationListener;
import javafx.beans.Observable;
import javafx.beans.value.ChangeListener;
import javafx.beans.value.ObservableValue;


public class Sources {

    public static Source<Void> fromInvalidations(Observable observable) {
        return new CombinedSourceBase<Void>() {
            private final InvalidationListener listener = obs -> emitValue(null);
            private final Subscription subscription = () -> observable.removeListener(listener);

            @Override
            protected Subscription subscribeToInputs() {
                observable.addListener(listener);
                return subscription;
            }

        };
    }

    public static <T> Source<T> fromChanges(ObservableValue<T> observable) {
        return new CombinedSourceBase<T>() {
            private final ChangeListener<T> listener = (obs, old, val) -> emitValue(val);
            private final Subscription subscription = () -> observable.removeListener(listener);

            @Override
            protected Subscription subscribeToInputs() {
                observable.addListener(listener);
                return subscription;
            }
        };
    }

    @SafeVarargs
    public static <U> Source<U> merge(Source<? extends U>... inputs) {
        return new CombinedSourceBase<U>() {
            @Override
            protected Subscription subscribeToInputs() {
                Subscription[] subs = new Subscription[inputs.length];
                for(int i = 0; i < inputs.length; ++i) {
                    subs[i] = inputs[i].subscribe(value -> emitValue(value));
                }
                return Subscription.multi(subs);
            }
        };
    }

    public static <T> Source<T> releaseOnImpulse(Source<?> impulse, Source<T> input) {
        return new CombinedSourceBase<T>() {
            private boolean hasValue = false;
            private T value = null;

            @Override
            protected Subscription subscribeToInputs() {
                Subscription s1 = input.subscribe(v -> {
                    hasValue = true;
                    value = v;
                });

                Subscription s2 = impulse.subscribe(i -> {
                    if(hasValue) {
                        emitValue(value);
                        hasValue = false;
                        value = null;
                    }
                });

                return Subscription.multi(s1, s2);
            }
        };
    }


    private static abstract class CombinedSourceBase<T> extends SourceBase<T> {
        private Subscription subscription = null;

        protected abstract Subscription subscribeToInputs();

        @Override
        protected final void firstSubscriber() {
            subscription = subscribeToInputs();
        }

        @Override
        protected final void noSubscribers() {
            subscription.unsubscribe();
            subscription = null;
        }
    }
}

@@ -196,16 +196,88 @@
 rces %7B%0A%0A
+    public static interface OnBuilder%3CT, R%3E %7B%0A        R on(T t);%0A    %7D%0A%0A
     publ
@@ -1786,32 +1786,180 @@
       %7D;%0A    %7D%0A%0A
+    public static %3CT%3E OnBuilder%3CSource%3C?%3E, Source%3CT%3E%3E release(Source%3CT%3E input) %7B%0A        return impulse -%3E releaseOnImpulse(impulse, input);%0A    %7D%0A%0A
     public stati

Support release(input).on(impulse) syntax.