openstack/net-virt/src/main/java/org/opendaylight/ovsdb/openstack/netvirt/impl/EventDispatcherImpl.java

   1 /*
   2  * Copyright (c) 2013, 2015 Red Hat, Inc. and others. All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.ovsdb.openstack.netvirt.impl;
  10
  11 import org.opendaylight.ovsdb.openstack.netvirt.AbstractEvent;
  12 import org.opendaylight.ovsdb.openstack.netvirt.AbstractHandler;
  13 import org.opendaylight.ovsdb.openstack.netvirt.ConfigInterface;
  14 import org.opendaylight.ovsdb.openstack.netvirt.api.Constants;
  15 import org.opendaylight.ovsdb.openstack.netvirt.api.EventDispatcher;
  16
  17 import org.osgi.framework.ServiceReference;
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 import java.util.concurrent.BlockingQueue;
  22 import java.util.concurrent.ExecutorService;
  23 import java.util.concurrent.Executors;
  24 import java.util.concurrent.LinkedBlockingQueue;
  25 import java.util.concurrent.TimeUnit;
  26
  27 public class EventDispatcherImpl implements EventDispatcher, ConfigInterface {
  28     private static final Logger LOG = LoggerFactory.getLogger(EventDispatcher.class);
  29     private ExecutorService eventHandler;
  30     private volatile BlockingQueue<AbstractEvent> events;
  31     private AbstractHandler[] handlers;
  32
  33     public EventDispatcherImpl() {
  34         events = new LinkedBlockingQueue<>();
  35         handlers = new AbstractHandler[AbstractEvent.HandlerType.size];
  36         eventHandler = Executors.newSingleThreadExecutor();
  37         start();
  38     }
  39
  40     void start() {
  41         eventHandler.submit(new Runnable()  {
  42             @Override
  43             public void run() {
  44                 Thread t = Thread.currentThread();
  45                 t.setName("EventDispatcherImpl");
  46                 LOG.info("EventDispatcherImpl: started {}", t.getName());
  47                 while (true) {
  48                     AbstractEvent ev;
  49                     try {
  50                         ev = events.take();
  51                     } catch (InterruptedException e) {
  52                         LOG.info("The event handler thread was interrupted, shutting down", e);
  53                         return;
  54                     }
  55                     try {
  56                         dispatchEvent(ev);
  57                     } catch (Exception e) {
  58                         LOG.error("Exception in dispatching event {}", ev.toString(), e);
  59                     }
  60                 }
  61             }
  62         });
  63         LOG.debug("event dispatcher is started");
  64     }
  65
  66     void stop() {
  67         // stop accepting new tasks
  68         eventHandler.shutdown();
  69         try {
  70             // Wait a while for existing tasks to terminate
  71             if (!eventHandler.awaitTermination(10, TimeUnit.SECONDS)) {
  72                 eventHandler.shutdownNow();
  73                 // Wait a while for tasks to respond to being cancelled
  74                 if (!eventHandler.awaitTermination(10, TimeUnit.SECONDS)) {
  75                     LOG.error("Dispatcher's event handler did not terminate");
  76                 }
  77             }
  78         } catch (InterruptedException e) {
  79             // (Re-)Cancel if current thread also interrupted
  80             eventHandler.shutdownNow();
  81             // Preserve interrupt status
  82             Thread.currentThread().interrupt();
  83         }
  84         LOG.debug("event dispatcher is stopped");
  85     }
  86
  87     private void dispatchEvent(AbstractEvent ev) {
  88         AbstractHandler handler = handlers[ev.getHandlerType().ordinal()];
  89         if (handler == null) {
  90             LOG.warn("event dispatcher found no handler for {}", ev);
  91             return;
  92         }
  93
  94         handler.processEvent(ev);
  95     }
  96
  97     public void eventHandlerAdded(final ServiceReference ref, AbstractHandler handler){
  98         Long pid = (Long) ref.getProperty(org.osgi.framework.Constants.SERVICE_ID);
  99         Object handlerTypeObject = ref.getProperty(Constants.EVENT_HANDLER_TYPE_PROPERTY);
 100         if (!(handlerTypeObject instanceof AbstractEvent.HandlerType)){
 101             // The exception should give us a stacktrace
 102             LOG.error("Abstract handler reg failed to provide a valid handler type: {} ref: {} handler: {}",
 103                     handlerTypeObject, ref.getClass().getName(), handler.getClass().getName(),
 104                     new IllegalArgumentException("Missing handler type"));
 105             return;
 106         }
 107         AbstractEvent.HandlerType handlerType = (AbstractEvent.HandlerType) handlerTypeObject;
 108         handlers[handlerType.ordinal()] = handler;
 109         LOG.info("eventHandlerAdded: handler: {}, pid: {}, type: {}",
 110                 handler.getClass().getName(), pid, handlerType);
 111     }
 112
 113     public void eventHandlerRemoved(final ServiceReference ref){
 114         Long pid = (Long) ref.getProperty(org.osgi.framework.Constants.SERVICE_ID);
 115         Object handlerTypeObject = ref.getProperty(Constants.EVENT_HANDLER_TYPE_PROPERTY);
 116         if (!(handlerTypeObject instanceof AbstractEvent.HandlerType)){
 117             LOG.error("Abstract handler unreg failed to provide a valid handler type {}", handlerTypeObject);
 118             return;
 119         }
 120         AbstractEvent.HandlerType handlerType = (AbstractEvent.HandlerType) handlerTypeObject;
 121         handlers[handlerType.ordinal()] = null;
 122         LOG.debug("Event handler for type {} unregistered pid {}", handlerType, pid);
 123     }
 124
 125     /**
 126      * Enqueue the event.
 127      *
 128      * @param event the {@link org.opendaylight.ovsdb.openstack.netvirt.AbstractEvent} event to be handled.
 129      */
 130     @Override
 131     public void enqueueEvent(AbstractEvent event) {
 132         if (event == null) {
 133             LOG.warn("enqueueEvent: event is null");
 134             return;
 135         }
 136
 137         try {
 138             events.put(event);
 139         } catch (InterruptedException e) {
 140             LOG.error("Thread was interrupted while trying to enqueue event ", e);
 141         }
 142     }
 143
 144     @Override
 145     public void setDependencies(ServiceReference serviceReference) {}
 146
 147     @Override
 148     public void setDependencies(Object impl) {}
 149 }