FairCallQueue extends AbstractQueue<E> implements BlockingQueue<E> { @Override public int drainTo(Collection<? super E> c, int maxElements) { final int permits = semaphore.drainPermits(); final int numElements = Math.min(maxElements, permits); int numRemaining = numElements; for (int i=0; numRemaining > 0 && i < queues.size(); i++) { numRemaining -= queues.get(i).drainTo(c, numRemaining); } int drained = numElements - numRemaining; if (permits > drained) { semaphore.release(permits - drained); } return drained; } FairCallQueue(int priorityLevels, int capacity, String ns,
      Configuration conf); @Override boolean add(E e); @Override void put(E e); @Override boolean offer(E e, long timeout, TimeUnit unit); @Override boolean offer(E e); @Override E take(); @Override E poll(long timeout, TimeUnit unit); @Override E poll(); @Override E peek(); @Override int size(); @Override Iterator<E> iterator(); @Override int drainTo(Collection<? super E> c, int maxElements); @Override int drainTo(Collection<? super E> c); @Override int remainingCapacity(); int[] getQueueSizes(); long[] getOverflowedCalls(); @VisibleForTesting void setMultiplexer(RpcMultiplexer newMux); @Deprecated
  static final int IPC_CALLQUEUE_PRIORITY_LEVELS_DEFAULT; @Deprecated
  static final String IPC_CALLQUEUE_PRIORITY_LEVELS_KEY; static final Logger LOG; }
@SuppressWarnings("deprecation") @Test public void testDrainTo() { Configuration conf = new Configuration(); conf.setInt("ns." + FairCallQueue.IPC_CALLQUEUE_PRIORITY_LEVELS_KEY, 2); FairCallQueue<Schedulable> fcq2 = new FairCallQueue<Schedulable>(2, 10, "ns", conf); for (int i = 0; i < 3; i++) { fcq.offer(mockCall("c")); } fcq.drainTo(fcq2); assertEquals(0, fcq.size()); assertEquals(3, fcq2.size()); }