}


    private void cleanup () {
        synchronized ( this.connections ) {
            SmbTransportImpl trans;
            while ( ( trans = this.toRemove.poll() ) != null ) {

                if ( log.isDebugEnabled() ) {
                    log.debug("Removing transport connection " + trans + " (" + System.identityHashCode(trans) + ")");
                }

    synchronized(this@RealWebSocket) {
      if (failed) {
        return false // Failed web socket.
      }

      writer = this.writer
      pong = pongQueue.poll()
      if (pong == null) {
        messageOrClose = messageAndCloseQueue.poll()
        if (messageOrClose is Close) {
          receivedCloseCode = this.receivedCloseCode
          receivedCloseReason = this.receivedCloseReason

      // If even the first one wasn't there, clean up so that the next test doesn't see an element.
      producerThread.cancel(true);
      producer.doneProducing.await();
      if (drained == 0) {
        q.poll(); // not necessarily there if producer was interrupted
      }
    }
  }

  public void testZeroElements() throws Exception {
    for (BlockingQueue<Object> q : blockingQueues()) {
      testZeroElements(q);

                // from execute().
                workerRunCount++;
                workerRunningState = RUNNING;
                hasSetRunning = true;
              }
            }
            task = queue.poll();
            if (task == null) {
              workerRunningState = IDLE;
              return;
            }
          }

        clearValueReferenceQueue();
      }
    }

    void clearKeyReferenceQueue() {
      while (keyReferenceQueue.poll() != null) {}
    }

    void clearValueReferenceQueue() {
      while (valueReferenceQueue.poll() != null) {}
    }

    // recency queue, shared by expiration and eviction

    /**

                final Map<String, Object> localDataMap =
                        dataMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
                String processingUrl = urlQueue.poll();
                processedUrls.add(processingUrl);
                if (deleteUrlList.contains(processingUrl)) {
                    deleteDocuments(); // delete before indexing
                }
                try {

      while (true) {
        Future<T> f = futureQueue.poll();
        if (f == null) {
          if (ntasks > 0) {
            --ntasks;
            futures.add(submitAndAddQueueListener(executorService, it.next(), futureQueue));
            ++active;
          } else if (active == 0) {
            break;
          } else if (timed) {
            f = futureQueue.poll(timeoutNanos, TimeUnit.NANOSECONDS);

    public void execute() {
      synchronized (runnables) {
        if (executed) {
          return;
        }
        executed = true;
      }

      while (!runnables.isEmpty()) {
        runnables.poll().execute();
      }
    }

    private static class RunnableExecutorPair {
      final Runnable runnable;
      final Executor executor;

      RunnableExecutorPair(Runnable runnable, Executor executor) {

        clearValueReferenceQueue();
      }
    }

    void clearKeyReferenceQueue() {
      while (keyReferenceQueue.poll() != null) {}
    }

    void clearValueReferenceQueue() {
      while (valueReferenceQueue.poll() != null) {}
    }

    // recency queue, shared by expiration and eviction

    /**

    // in the array. We could skip this if we decide we don't care about holding on to Reference
    // objects indefinitely.
    private void drainQueue() {
      Reference<? extends L> ref;
      while ((ref = queue.poll()) != null) {
        // We only ever register ArrayReferences with the queue so this is always safe.
        ArrayReference<? extends L> arrayRef = (ArrayReference<? extends L>) ref;