package org.reactfx;

import java.util.function.Consumer;

public abstract class EventStreamBase<T> implements EventStream<T> {

    private ListHelper<Consumer<? super T>> subscribers = null;

    protected void emit(T value) {
        ListHelper.forEach(subscribers, s -> s.accept(value));
    }

    /**
     * Called when the number of subscribers goes from 0 to 1.
     * Overriding this method is a convenient way for subclasses
     * to handle this event.
     */
    protected void firstSubscriber() {
        // default implementation is empty
    }

    /**
     * Called when the number of subscribers goes down to 0.
     * Overriding this method is a convenient way for subclasses
     * to handle this event.
     */
    protected void noSubscribers() {
        // default implementation is empty
    }

    @Override
    public Subscription subscribe(Consumer<? super T> consumer) {
        subscribers = ListHelper.add(subscribers, consumer);
        if(ListHelper.size(subscribers) == 1) {
            firstSubscriber();
        }
        return () -> unsubscribe(consumer);
    }

    private void unsubscribe(Consumer<? super T> consumer) {
        subscribers = ListHelper.remove(subscribers, consumer);
        if(ListHelper.isEmpty(subscribers)) {
            noSubscribers();
        }
    }
}

@@ -467,45 +467,454 @@
    *
-/%0A    protected void firstSubscriber(
+%0A     * %3Cp%3EThis method is called after the %7B@link #newSubscriber(Consumer)%7D%0A     * method.%3C/p%3E%0A     */%0A    protected void firstSubscriber() %7B%0A        // default implementation is empty%0A    %7D%0A%0A    /**%0A     * Called for each new subscriber.%0A     * Overriding this method is a convenient way for subclasses%0A     * to handle this event, for example to publish some initial events.%0A     */%0A    protected void newSubscriber(Consumer%3C? super T%3E consumer
 ) %7B%0A
@@ -1354,32 +1354,65 @@
 ers, consumer);%0A
+        newSubscriber(consumer);%0A
         if(ListH

Add newSubscriber() hook.