3
\$\begingroup\$

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();
    }
};
\$\endgroup\$
1
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ The purpose of multithreading is to keep from blocking the caller, so as long as I have some form of queue to manage execution order that should be sorted out. I agree with the separation of concerns between Downloader and Response, but I'm not sure what you mean by the Download class: a Response represents a download in progress. \$\endgroup\$ – ndm13 Jul 14 '18 at 15:46
  • \$\begingroup\$ If one could implement multiple generic instances of the same interface, we could get around the SRP problem if I understand it correctly. As it stands, the most elegant solution I can think of is what I've implemented (maybe with default no-op methods) or some form of Map<Class<T>,Set<Listener<T>>> that each implementer would have to maintain and check for consistency. \$\endgroup\$ – ndm13 Jul 14 '18 at 15:52
  • \$\begingroup\$ The point is: The ForkJoinPool is a pool and not a queue. So there is no chance to ensure the execution order of the notifications. With this implementation a "downloadUpdate" may come before a "downloadStarted". \$\endgroup\$ – oopexpert Jul 14 '18 at 16:51
  • \$\begingroup\$ I understand. So if I added a queue in some form to ensure ordering, this would be alleviated, correct? The problem being that there's no global pool that supports this, so there would have to be an implementation in each class that supplies an instance-level Executor. \$\endgroup\$ – ndm13 Jul 14 '18 at 17:52
  • \$\begingroup\$ If you can ensure that each notification for an observer is executed with the same executor and furthermore is being queued if the executor is busy then this would be a way to go. The executor must be bound to the observer as long as at least one notification is processed. You have to have something between the Observable and the ForkJoinPool that will manage it. \$\endgroup\$ – oopexpert Jul 14 '18 at 21:03

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.