Adding event sink & dispatcher abstractions.

Increased strength of the provider & service types.

+package org.onlab.onos.event;
+
+import com.google.common.collect.ImmutableSet;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Base implementation of event sink broker.
+ */
+public class AbstractEventSinkBroker implements EventSinkBroker {
+
+    private final Map<Class<? extends Event>, EventSink<? extends Event>> sinks =
+            new ConcurrentHashMap<>();
+
+    @Override
+    public <E extends Event> void addSink(Class<E> eventClass, EventSink<E> sink) {
+        checkNotNull(eventClass, "Event class cannot be null");
+        checkNotNull(sink, "Event sink cannot be null");
+        checkArgument(!sinks.containsKey(eventClass),
+                      "Event sink already registered for %s", eventClass.getName());
+        sinks.put(eventClass, sink);
+    }
+
+    @Override
+    public <E extends Event> void removeSink(Class<E> eventClass) {
+        checkNotNull(eventClass, "Event class cannot be null");
+        checkArgument(sinks.remove(eventClass) != null,
+                      "Event sink not registered for %s", eventClass.getName());
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public <E extends Event> EventSink<E> getSink(Class<E> eventClass) {
+        return (EventSink<E>) sinks.get(eventClass);
+    }
+
+    @Override
+    public Set<Class<? extends Event>> getSinks() {
+        return ImmutableSet.copyOf(sinks.keySet());
+    }
+}

+package org.onlab.onos.event;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Base implementation of a manager capable of tracking listeners and
+ * dispatching events to them.
+ */
+public class AbstractListenerManager<E extends Event, L extends EventListener<E>>
+        implements EventSink<E> {
+
+    protected Logger log = LoggerFactory.getLogger(AbstractListenerManager.class);
+
+    private final Set<L> listeners = new CopyOnWriteArraySet<>();
+
+    /**
+     * Adds the specified listener.
+     *
+     * @param listener listener to be added
+     */
+    public void addListener(L listener) {
+        checkNotNull(listener, "Listener cannot be null");
+        listeners.add(listener);
+    }
+
+    /**
+     * Removes the specified listener.
+     *
+     * @param listener listener to be removed
+     */
+    public void removeListener(L listener) {
+        checkNotNull(listener, "Listener cannot be null");
+        checkArgument(listeners.remove(listener), "Listener not registered");
+    }
+
+    @Override
+    public void process(E event) {
+        for (L listener : listeners) {
+            try {
+                listener.event(event);
+            } catch (Throwable error) {
+                reportProblem(event, error);
+            }
+        }
+    }
+
+    @Override
+    public void reportProblem(E event, Throwable error) {
+        log.warn("Exception encountered while processing event " + event, error);
+    }
+
+}

+package org.onlab.onos.event;
+
+/**
+ * Abstraction of a mechanism capable of accepting and dispatching events.
+ * Whether the events are accepted and the dispatched synchronously or
+ * asynchronously is unspecified.
+ */
+public interface EventDispatcher<E extends Event> {
+
+    /**
+     * Posts the specified event for dispatching.
+     *
+     * @param event event to be posted
+     */
+    void post(E event);
+
+}

+package org.onlab.onos.event;
+
+/**
+ * Abstraction of an event sink capable of processing the specified event types.
+ */
+public interface EventSink<E extends Event> {
+
+    /**
+     * Processes the specified event.
+     *
+     * @param event event to be processed
+     */
+    void process(E event);
+
+    /**
+     * Reports a problem encountered while processing an event.
+     *
+     * @param event event being processed
+     * @param error error encountered while processing
+     */
+    void reportProblem(E event, Throwable error);
+
+}

+package org.onlab.onos.event;
+
+import java.util.Set;
+
+/**
+ * Abstraction of an event sink broker capable of tracking sinks based on
+ * their event class.
+ */
+public interface EventSinkBroker {
+
+    /**
+     * Adds the specified sink for the given event class.
+     *
+     * @param eventClass event class
+     * @param sink       event sink
+     * @param <E>        type of event
+     */
+    <E extends Event> void addSink(Class<E> eventClass, EventSink<E> sink);
+
+    /**
+     * Removes the sink associated with the given event class.
+     *
+     * @param eventClass event class
+     * @param <E>        type of event
+     */
+    <E extends Event> void removeSink(Class<E> eventClass);
+
+    /**
+     * Returns the event sink associated with the specified event class.
+     *
+     * @param eventClass event class
+     * @param <E>        type of event
+     * @return event sink or null if none found
+     */
+    <E extends Event> EventSink<E> getSink(Class<E> eventClass);
+
+    /**
+     * Returns the set of all event classes for which sinks are presently
+     * registered.
+     *
+     * @return set of event classes
+     */
+    Set<Class<? extends Event>> getSinks();
+
+}

@@ -9,7 +9,7 @@ import java.util.List;
  * Service through which device providers can inject device information into
  * Service through which device providers can inject device information into
  * the core.
  * the core.
  */
  */
-public interface DeviceProviderService extends ProviderService {
+public interface DeviceProviderService extends ProviderService<DeviceProvider> {
 
 
     // TODO: define suspend and remove actions on the mezzanine administrative API
     // TODO: define suspend and remove actions on the mezzanine administrative API
 
 

@@ -6,7 +6,7 @@ import org.onlab.onos.net.provider.ProviderService;
  * Service through which flowrule providers can inject flowrule information into
  * Service through which flowrule providers can inject flowrule information into
  * the core.
  * the core.
  */
  */
-public interface FlowRuleProviderService extends ProviderService {
+public interface FlowRuleProviderService extends ProviderService<FlowRuleProvider> {
 
 
     /**
     /**
      * Signals that a flow that was previously installed has been removed.
      * Signals that a flow that was previously installed has been removed.

@@ -5,7 +5,7 @@ import org.onlab.onos.net.provider.ProviderService;
 /**
 /**
  * Means of conveying host information to the core.
  * Means of conveying host information to the core.
  */
  */
-public interface HostProviderService extends ProviderService {
+public interface HostProviderService extends ProviderService<HostProvider> {
 
 
     /**
     /**
      * Notifies the core when a host has been detected on a network along with
      * Notifies the core when a host has been detected on a network along with

@@ -5,7 +5,7 @@ import org.onlab.onos.net.provider.ProviderService;
 /**
 /**
  * Means for injecting link information into the core.
  * Means for injecting link information into the core.
  */
  */
-public interface LinkProviderService extends ProviderService {
+public interface LinkProviderService extends ProviderService<LinkProvider> {
 
 
     /**
     /**
      * Signals that an infrastructure link has been connected.
      * Signals that an infrastructure link has been connected.

@@ -12,7 +12,7 @@ import static com.google.common.base.Preconditions.checkNotNull;
  * @param <P> type of the information provider
  * @param <P> type of the information provider
  * @param <S> type of the provider service
  * @param <S> type of the provider service
  */
  */
-public abstract class AbstractProviderBroker<P extends Provider, S extends ProviderService>
+public abstract class AbstractProviderBroker<P extends Provider, S extends ProviderService<P>>
         implements ProviderBroker<P, S> {
         implements ProviderBroker<P, S> {
 
 
     private final Map<ProviderId, S> services = new HashMap<>();
     private final Map<ProviderId, S> services = new HashMap<>();

@@ -5,7 +5,7 @@ import org.onlab.onos.net.provider.ProviderService;
 /**
 /**
  * Means for injecting topology information into the core.
  * Means for injecting topology information into the core.
  */
  */
-public interface TopologyProviderService extends ProviderService {
+public interface TopologyProviderService extends ProviderService<TopologyProvider> {
 
 
     // What can be conveyed in a topology that isn't by individual
     // What can be conveyed in a topology that isn't by individual
     // providers?
     // providers?