openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/queue/QueueKeeperLightImpl.java

   1 /**
   2  * Copyright (c) 2013 Cisco Systems, 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 package org.opendaylight.openflowplugin.openflow.md.queue;
   9
  10 import java.util.ArrayList;
  11 import java.util.Collection;
  12 import java.util.List;
  13 import java.util.Map;
  14 import java.util.concurrent.BlockingQueue;
  15 import java.util.concurrent.LinkedBlockingQueue;
  16 import java.util.concurrent.ScheduledThreadPoolExecutor;
  17
  18 import org.opendaylight.openflowplugin.openflow.md.core.ConnectionConductor;
  19 import org.opendaylight.openflowplugin.openflow.md.core.IMDMessageTranslator;
  20 import org.opendaylight.openflowplugin.openflow.md.core.SwitchConnectionDistinguisher;
  21 import org.opendaylight.openflowplugin.openflow.md.core.TranslatorKey;
  22 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  23 import org.opendaylight.yangtools.yang.binding.DataObject;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 /**
  28  * @author mirehak
  29  */
  30 public class QueueKeeperLightImpl implements QueueKeeper<OfHeader, DataObject> {
  31
  32     private static final Logger LOG = LoggerFactory
  33             .getLogger(QueueKeeperLightImpl.class);
  34
  35     private Map<Class<? extends DataObject>, Collection<PopListener<DataObject>>> popListenersMapping;
  36     private BlockingQueue<TicketResult<DataObject>> processQueue;
  37     private ScheduledThreadPoolExecutor pool;
  38     private int poolSize = 10;
  39     private Map<TranslatorKey, Collection<IMDMessageTranslator<OfHeader, List<DataObject>>>> translatorMapping;
  40     private TicketProcessorFactory<OfHeader, DataObject> ticketProcessorFactory;
  41     private MessageSpy<OfHeader, DataObject> messageSpy;
  42
  43     private VersionExtractor<OfHeader> versionExtractor = new VersionExtractor<OfHeader>() {
  44         @Override
  45         public Short extractVersion(OfHeader message) {
  46             return message.getVersion();
  47         }
  48     };
  49
  50     private RegisteredTypeExtractor<OfHeader> registeredSrcTypeExtractor =
  51             new RegisteredTypeExtractor<OfHeader>() {
  52         @SuppressWarnings("unchecked")
  53         @Override
  54         public Class<? extends OfHeader> extractRegisteredType(
  55                 OfHeader message) {
  56             return (Class<? extends OfHeader>) message.getImplementedInterface();
  57         }
  58     };
  59
  60     private RegisteredTypeExtractor<DataObject> registeredOutTypeExtractor =
  61             new RegisteredTypeExtractor<DataObject>() {
  62         @SuppressWarnings("unchecked")
  63         @Override
  64         public Class<? extends DataObject> extractRegisteredType(
  65                 DataObject message) {
  66             return (Class<? extends DataObject>) message.getImplementedInterface();
  67         }
  68     };
  69
  70     /**
  71      * prepare queue
  72      */
  73     public void init() {
  74         processQueue = new LinkedBlockingQueue<>(100);
  75         pool = new ScheduledThreadPoolExecutor(poolSize);
  76
  77         ticketProcessorFactory = new TicketProcessorFactory<>();
  78         ticketProcessorFactory.setRegisteredTypeExtractor(registeredSrcTypeExtractor);
  79         ticketProcessorFactory.setTranslatorMapping(translatorMapping);
  80         ticketProcessorFactory.setVersionExtractor(versionExtractor);
  81         ticketProcessorFactory.setSpy(messageSpy);
  82
  83         TicketFinisher<DataObject> finisher = new TicketFinisher<>(
  84                 processQueue, popListenersMapping, registeredOutTypeExtractor);
  85         new Thread(finisher).start();
  86     }
  87
  88     /**
  89      * stop processing queue
  90      */
  91     public void shutdown() {
  92         pool.shutdown();
  93     }
  94
  95     @Override
  96     public void push(OfHeader message, ConnectionConductor conductor) {
  97         push(message,conductor,QueueKeeper.QueueType.DEFAULT);
  98     }
  99
 100     @Override
 101     public void push(OfHeader message, ConnectionConductor conductor, QueueType queueType) {
 102         if(queueType == QueueKeeper.QueueType.DEFAULT) {
 103             TicketImpl<OfHeader, DataObject> ticket = new TicketImpl<>();
 104             ticket.setConductor(conductor);
 105             ticket.setMessage(message);
 106             LOG.debug("ticket scheduling: {}, ticket: {}",
 107                     message.getImplementedInterface().getSimpleName(), System.identityHashCode(ticket));
 108             //TODO: block if queue limit reached
 109             processQueue.add(ticket);
 110             scheduleTicket(ticket);
 111         } else if (queueType == QueueKeeper.QueueType.UNORDERED){
 112             List<DataObject> processedMessages = translate(message,conductor);
 113             pop(processedMessages,conductor);
 114         }
 115     }
 116
 117     /**
 118      * @param ticket
 119      */
 120     private void scheduleTicket(Ticket<OfHeader, DataObject> ticket) {
 121         Runnable ticketProcessor = ticketProcessorFactory.createProcessor(ticket);
 122         pool.execute(ticketProcessor);
 123     }
 124
 125     /**
 126      * @param poolSize the poolSize to set
 127      */
 128     public void setPoolSize(int poolSize) {
 129         this.poolSize = poolSize;
 130     }
 131
 132     @Override
 133     public void setTranslatorMapping(
 134             Map<TranslatorKey, Collection<IMDMessageTranslator<OfHeader, List<DataObject>>>> translatorMapping) {
 135         this.translatorMapping = translatorMapping;
 136     }
 137
 138     @Override
 139     public void setPopListenersMapping(
 140             Map<Class<? extends DataObject>, Collection<PopListener<DataObject>>> popListenersMapping) {
 141         this.popListenersMapping = popListenersMapping;
 142     }
 143
 144     /**
 145      * @param messageSpy the messageSpy to set
 146      */
 147     public void setMessageSpy(MessageSpy<OfHeader, DataObject> messageSpy) {
 148         this.messageSpy = messageSpy;
 149     }
 150
 151     private List<DataObject> translate(OfHeader message, ConnectionConductor conductor) {
 152         List<DataObject> result = new ArrayList<>();
 153         Class<? extends OfHeader> messageType = registeredSrcTypeExtractor.extractRegisteredType(message);
 154         Collection<IMDMessageTranslator<OfHeader, List<DataObject>>> translators = null;
 155         LOG.debug("translating message: {}", messageType.getSimpleName());
 156
 157         Short version = versionExtractor.extractVersion(message);
 158         if (version == null) {
 159            throw new IllegalArgumentException("version is NULL");
 160         }
 161         TranslatorKey tKey = new TranslatorKey(version, messageType.getName());
 162         translators = translatorMapping.get(tKey);
 163
 164         LOG.debug("translatorKey: {} + {}", version, messageType.getName());
 165
 166         if (translators != null) {
 167             for (IMDMessageTranslator<OfHeader, List<DataObject>> translator : translators) {
 168                 SwitchConnectionDistinguisher cookie = null;
 169                 // Pass cookie only for PACKT_IN
 170                 if (messageType.equals("PacketInMessage.class")) {
 171                     cookie = conductor.getAuxiliaryKey();
 172                 }
 173                 List<DataObject> translatorOutput = translator.translate(cookie, conductor.getSessionContext(), message);
 174                 if(translatorOutput != null) {
 175                     result.addAll(translator.translate(cookie, conductor.getSessionContext(), message));
 176                 }
 177             }
 178             if (messageSpy != null) {
 179                 messageSpy.spyIn(message);
 180                 messageSpy.spyOut(result);
 181             }
 182         } else {
 183             LOG.warn("No translators for this message Type: {}", messageType);
 184         }
 185         return result;
 186     }
 187
 188     private void pop(List<DataObject> processedMessages,ConnectionConductor conductor) {
 189         for (DataObject msg : processedMessages) {
 190             Class<? extends Object> registeredType =
 191                     registeredOutTypeExtractor.extractRegisteredType(msg);
 192             Collection<PopListener<DataObject>> popListeners = popListenersMapping.get(registeredType);
 193             if (popListeners == null) {
 194                 LOG.warn("no popListener registered for type {}"+registeredType);
 195             } else {
 196                 for (PopListener<DataObject> consumer : popListeners) {
 197                     consumer.onPop(msg);
 198                 }
 199             }
 200         }
 201     }
 202 }