openflowplugin-api/src/main/java/org/opendaylight/openflowplugin/api/openflow/md/util/PollableQueuesZipper.java

   1 /**
   2  * Copyright (c) 2014 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.api.openflow.md.util;
   9
  10 import java.util.ArrayList;
  11 import java.util.Iterator;
  12 import java.util.List;
  13 import java.util.Queue;
  14
  15 import com.google.common.collect.Iterators;
  16
  17 /**
  18  * Zipper groups together a list of queues and exposes one poll method. Polling iterates through
  19  * all groups and returns first not-null result of poll method on each queue. If after polling each
  20  * grouped queue for one time there is still null result, poll will return null.
  21  * <br>
  22  * Iterating keeps last position so this polling is supposed to be fairly distributed.
  23  *
  24  * @param <T> common item type of zipped queues
  25  */
  26 public class PollableQueuesZipper<T> {
  27
  28     private List<Queue<T>> sources;
  29     private Iterator<Queue<T>> cursor;
  30
  31     /**
  32      * default ctor
  33      */
  34     public PollableQueuesZipper() {
  35         sources = new ArrayList<>();
  36     }
  37
  38     /**
  39      * Add all member queues before first invocation of {@link PollableQueuesZipper#poll()}
  40      * @param queue to be added to group
  41      */
  42     public void addSource(Queue<T> queue) {
  43         sources.add(queue);
  44     }
  45
  46     /**
  47      * @return next common product of polling member groups
  48      */
  49     public T poll() {
  50         T item = null;
  51         if (cursor == null) {
  52             cursor = Iterators.cycle(sources);
  53         }
  54
  55         Queue<T> queue;
  56         for (int i = 0; i < sources.size(); i++) {
  57             queue = cursor.next();
  58             item = queue.poll();
  59             if (item != null) {
  60                 break;
  61             }
  62         }
  63
  64         return item;
  65     }
  66 }