openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/queue/QueueKeeperFactory.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 org.opendaylight.openflowplugin.api.openflow.md.queue.MessageSourcePollRegistrator;
  11 import org.opendaylight.openflowplugin.api.openflow.md.queue.QueueKeeper;
  12 import org.opendaylight.openflowplugin.api.openflow.md.queue.WaterMarkListener;
  13 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  14
  15 /**
  16  * factory for
  17  * {@link org.opendaylight.openflowplugin.api.openflow.md.queue.QueueKeeper}
  18  * implementations
  19  */
  20 public abstract class QueueKeeperFactory {
  21
  22     /**
  23      * @param sourceRegistrator
  24      * @param capacity
  25      *            blocking queue capacity
  26      * @param waterMarkListener
  27      * @return fair reading implementation of
  28      *         {@link org.opendaylight.openflowplugin.api.openflow.md.queue.QueueKeeper}
  29      *         (not registered = not started yet)
  30      */
  31     public static QueueKeeper<OfHeader> createFairQueueKeeper(
  32             MessageSourcePollRegistrator<QueueKeeper<OfHeader>> sourceRegistrator,
  33             int capacity, WaterMarkListener waterMarkListener) {
  34         QueueKeeperFairImpl queueKeeper = new QueueKeeperFairImpl();
  35         queueKeeper.setCapacity(capacity);
  36         queueKeeper.setHarvesterHandle(sourceRegistrator.getHarvesterHandle());
  37         queueKeeper.setWaterMarkListener(waterMarkListener);
  38         queueKeeper.init();
  39
  40         return queueKeeper;
  41     }
  42
  43     /**
  44      * register queue by harvester, start processing it. Use
  45      * {@link QueueKeeperFairImpl#close()} to kill the queue and stop
  46      * processing.
  47      *
  48      * @param sourceRegistrator
  49      * @param queueKeeper
  50      */
  51     public static <V> void plugQueue(
  52             MessageSourcePollRegistrator<QueueKeeper<V>> sourceRegistrator,
  53             QueueKeeper<V> queueKeeper) {
  54         AutoCloseable registration = sourceRegistrator
  55                 .registerMessageSource(queueKeeper);
  56         queueKeeper.setPollRegistration(registration);
  57         sourceRegistrator.getHarvesterHandle().ping();
  58     }
  59 }