package org.reactfx;

@FunctionalInterface
public interface Subscription {
    void unsubscribe();

    /**
     * Returns a new aggregate subscription whose {@link #unsubscribe()}
     * method calls {@code unsubscribe()} on both this subscription and
     * {@code other} subscription.
     */
    default Subscription and(Subscription other) {
        return new BiSubscription(this, other);
    }

    static final Subscription EMPTY = () -> {};

    /**
     * Returns a new aggregate subscription whose {@link #unsubscribe()}
     * method calls {@code unsubscribe()} on all arguments to this method.
     */
    static Subscription multi(Subscription... subs) {
        switch(subs.length) {
            case 0: return EMPTY;
            case 1: return subs[0];
            case 2: return new BiSubscription(subs[0], subs[1]);
            default: return new MultiSubscription(subs);
        }
    }
}

class BiSubscription implements Subscription {
    private final Subscription s1;
    private final Subscription s2;

    public BiSubscription(Subscription s1, Subscription s2) {
        this.s1 = s1;
        this.s2 = s2;
    }

    @Override
    public void unsubscribe() {
        s1.unsubscribe();
        s2.unsubscribe();
    }
}

class MultiSubscription implements Subscription {
    private final Subscription[] subscriptions;

    public MultiSubscription(Subscription... subscriptions) {
        this.subscriptions = subscriptions;
    }

    @Override
    public void unsubscribe() {
        for(Subscription s: subscriptions) {
            s.unsubscribe();
        }
    }
}
@@ -15,16 +15,143 @@
 actfx;%0A%0A
+import java.util.HashMap;%0Aimport java.util.Map;%0Aimport java.util.function.Function;%0A%0Aimport javafx.collections.ObservableSet;%0A%0A
 @Functio
@@ -1022,24 +1022,1548 @@
      %7D%0A    %7D
+%0A%0A    /**%0A     * Dynamically subscribes to all elements of the given observable set.%0A     * When an element is added to the set, it is automatically subscribed to.%0A     * When an element is removed from the set, it is automatically unsubscribed%0A     * from.%0A     * @param elems observable set of elements that will be subscribed to%0A     * @param f function to subscribe to an element of the set.%0A     * @return An aggregate subscription that tracks elementary subscriptions.%0A     * When the returned subscription is unsubscribed, all elementary%0A     * subscriptions are unsubscribed as well, and no new elementary%0A     * subscriptions will be created.%0A     */%0A    static %3CT%3E Subscription multi(%0A            ObservableSet%3CT%3E elems,%0A            Function%3C? super T, ? extends Subscription%3E f) %7B%0A%0A        Map%3CT, Subscription%3E elemSubs = new HashMap%3C%3E();%0A        elems.forEach(t -%3E elemSubs.put(t, f.apply(t)));%0A%0A        Subscription setSub = EventStreams.changesOf(elems).subscribe(ch -%3E %7B%0A            if(ch.wasRemoved()) %7B%0A                Subscription sub = elemSubs.remove(ch.getElementRemoved());%0A                assert sub != null;%0A                sub.unsubscribe();%0A            %7D%0A            if(ch.wasAdded()) %7B%0A                T elem = ch.getElementAdded();%0A                assert !elemSubs.containsKey(elem);%0A                elemSubs.put(elem, f.apply(elem));%0A            %7D%0A        %7D);%0A%0A        return () -%3E %7B%0A            setSub.unsubscribe();%0A            elemSubs.forEach((t, sub) -%3E sub.unsubscribe());%0A        %7D;%0A    %7D
 %0A%7D%0A%0Aclass Bi

Add Subscription.multi(ObsSet<T>, T->Subscription)