Lightweight Java Event Handling Interface

Question

I've been looking for a simple way to manage event passing between classes and, like many others, have found the whole addXxxxListener paradigm to be tedious and overly verbose. I wrote up an interface with several default methods that seems to be working out fairly well for me, but before I decide to lean too heavily on it I felt it was a good idea to get some feedback.

My goals:

• Simplicity - Easy to implement and cover a vast majority of use cases
• Speed - Don't block the calling thread any more than is necessary

Basic code below. The Dispatcher interface is the only thing under review: the rest is heavily simplified code used to form a sample use case.

The Main Interface - Dispatcher

import java.util.EventListener;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Consumer;

public interface Dispatcher<L extends EventListener>{
    /**
     * A set that contains all the event listeners attached
     * to this object.
     *
     * @return  A {@link Set} of event listeners ({@link L})
     */
    Set<L> getEventListeners();

    /**
     * The {@link Executor} that handles event propagation.
     *
     * Defaults to {@link ForkJoinPool#commonPool()}.
     *
     * @see     #dispatch(Consumer) for more information on
     *          how events are handled.
     * @return  The Executor that will handle the events.
     */
    default Executor getEventExecutor(){
        return ForkJoinPool.commonPool();
    }

    /**
     * Adds an event listener to this object.  All events
     * announced to this object will be received by the
     * provided listener ({@link L}) until it is
     * {@link #removeEventListener(EventListener) removed}.
     *
     * Defaults to {@link #getEventListeners()}.{@link Set#add(Object) add(listener)}.
     *
     * @param listener
     *          The event listener to add.
     */
    default void addEventListener(L listener){
        getEventListeners().add(listener);
    }

    /**
     * Removes an event listener from this object.  Any
     * events received after this action occurs will not be
     * forwarded to this listener on a best-effort basis.
     *
     * Defaults to {@link #getEventListeners()}.{@link Set#remove(Object) remove(listener)}.
     *
     * @param listener
     *          The event listener to remove.
     */
    default void removeEventListener(L listener){
        getEventListeners().remove(listener);
    }

    /**
     * Forwards an event to all listeners currently in the
     * Set received from {@link #getEventListeners()}.  The
     * design intent is that, for a listener in the form
     * {@code void onUpdate(Data data);}, this method can
     * be invoked on all listeners by calling
     * {@code dispatch(l->l.onUpdate(data));}.
     *
     * The default implementation submits a new call to
     * {@link Consumer#accept(Object)} to the {@link Executor}
     * provided by {@link #getEventExecutor()} for each
     * event listener ({@link L}) in the backing collection
     * ({@link #getEventListeners()}).
     *
     * @param action
     */
    default void dispatch(Consumer<L> action){
        getEventListeners().forEach(l->getEventExecutor().execute(()->action.accept(l)));
    }
}

Setup

import java.io.IOException;
import java.net.URL;
import java.util.EventListener;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;

import Dispatcher;

public abstract class Downloader implements Dispatcher<Downloader.Listener>{
    private Set<Listener> listeners = new CopyOnWriteArraySet<>();

    @Override
    public Set<Listener> getEventListeners(){
        return listeners;
    }

    public abstract Response download(URL url) throws IOException;

    public interface Listener extends EventListener{
        void onDownloadStarted(Response response);
        void onDownloadUpdate(Response response);
        void onDownloadComplete(Response response);
        void onDownloadError(Response response, Exception exception);
    }
}

Dispatching

public Response download(URL url) throws IOException{
    // ...
    Response response = new Response(...);
    dispatch(l->l.onDownloadStarted(response));

    if(isInfiniteStream(connection)){
        response.completeInfinite(...);
        dispatch(l->l.onDownloadComplete(response));
        return response;
    }

    try(BufferedInputStream inputStream = ...){
        // ...
        while((read = inputStream.read(buffer)) > -1){
            response.update(...);
            dispatch(l->l.onDownloadUpdate(response));
        }
        response.complete(...);
        dispatch(l->l.onDownloadComplete(response));
        return response;
    }catch(IOException exception){
        dispatch(l->l.onDownloadError(response, exception));
        throw exception;
    }
}

Listening

static final Downloader.Listener downloadConsoleLogger = new Downloader.Listener(){
    @Override
    public void onDownloadStarted(Response response){
        System.out.println("Download started: " + response.getURL().toExternalForm());
    }

    @Override
    public void onDownloadUpdate(Response response){
        System.out.println("Update from " + response.getURL().toExternalForm() + ":\n\t" +
                "cs:" + response.getCurrentSpeed() + " as:" + response.getAverageSpeed() +
                "cb:" + response.getDownloadedSize() + " ab:" + response.getEstimatedSize());
    }

    @Override
    public void onDownloadComplete(Response response){
        System.out.println("Download complete: " + response.getURL().toExternalForm() +
                "\n\t as:" + response.getAverageSpeed() +
                " tb:" + response.getDownloadedSize());
    }

    @Override
    public void onDownloadError(Response response, Exception exception){
        System.err.println("Download failed: " + response.getURL().toExternalForm());
        exception.printStackTrace();
    }
};

Answer 1

You should omit the Executor as this will probably execute the notification within a separate thread. I guess this was your intention. But the Listener interface as you specified it has call dependencies. First you expect a "downloadStarted", then several "updates" and finally "downloadComplete". As you do not know when the Thread executions really starts the order of the notifications ist not ensured anymore. To execute publish notifications in parallel they mustn't have an ensured order of appearence. Where a sequence has to be ensured there cannot be parallelism.

You may do so without the Executor. But you restrict yourself to exactly ONE aspect to listen for per observable object. As several Listener types on one observable object tend to violate the SRP (Single Responsible Principle) this is enforcing you to do the right thing most of the time. But think about Listeners that have the same purpose but deal with more details.

I have to say that this way of declaring a generic interface for dispatching listener events is quite elegant aside the mentioned drawback. It at least reduced the elements to be implemented within the Observable to a minimum. So far as JAVA language mechanisms allow.

But aside this I want to say something to the code "not under review" ;-):

One thing is the assumed thread-safety. Yes, the CopyOnWriteArraySet makes some part technically thread-safe. But I woud have at least semantic questions related to the structure.

As you are expecting this to be working within a multithread environment you may experiencesome issus that are related to the call order of the observer notifications. CopyOnWriteArraySet will ensure immutability on the iterator once retreived as the documentation says:

Traversal via iterators is fast and cannot encounter interference from other threads. Iterators rely on unchanging snapshots of the array at the time the iterators were constructed.

But you may experience notifications on Listeners like "update" without having received a download started before. This may be possible but you have to ensure, that any "update" notification contains everything needed to let the Listener prepare its initial state or you go with this:

I suggest to have an intial notification once the Listener is registered properly that will provide the current state af all downloads in progress so the Listener is able to adapt his initial state in a dedicated way.

Additionally you should only provide immutable objects to Listeners. In your case the Listener will not be hindered to update the Response by itself and mess up the state for all other Listeners.

You also have another semantic issue. You are registering at a "Downloader" but you want to observe "Downloads". So you should encapsulate each download execution in a separate object and make the listener registration available there.

You can also observe a "Downloader"-object. But this object type may have some different information to publish like: downloadCreated, downloadDisposed.