code: /*
   * Copyright (c) 2019 the Eclipse Milo Authors
   *
   * This program and the accompanying materials are made
   * available under the terms of the Eclipse Public License 2.0
   * which is available at https://www.eclipse.org/legal/epl-2.0/
   *
   * SPDX-License-Identifier: EPL-2.0
   */
  
  package org.eclipse.milo.opcua.stack.core.util;
  
  import java.util.ArrayDeque;
  import java.util.concurrent.Executor;
  
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Queues up submitted {@link java.lang.Runnable}s and executes on an {@link java.util.concurrent.Executor}, with
   * optional concurrency.
   * <p>
   * When {@code concurrency = 1} (the default) submitted tasks are guaranteed to run serially and in the order submitted.
   * <p>
   * When {@code concurrency > 1} there are no guarantees beyond the fact that tasks are still pulled from a queue to be
   * executed.
   */
  public class ExecutionQueue {
  
      private final Logger log = LoggerFactory.getLogger(getClass());
  
      private final Object queueLock = new Object();
      private final ArrayDeque<Runnable> queue = new ArrayDeque<>();
  
      private int pending = 0;
      private boolean paused = false;
  
      private final Executor executor;
      private final int concurrencyLimit;
  
      public ExecutionQueue(Executor executor) {
          this(executor, 1);
      }
  
      public ExecutionQueue(Executor executor, int concurrencyLimit) {
          this.executor = executor;
          this.concurrencyLimit = concurrencyLimit;
      }
  
      /**
       * Submit a {@link Runnable} to be executed.
       *
       * @param runnable the {@link Runnable} to be executed.
       */
      public void submit(Runnable runnable) {
          synchronized (queueLock) {
              queue.add(runnable);
  
              maybePollAndExecute();
          }
      }
  
      /**
       * Submit a {@link Runnable} to be executed at the head of the queue.
       *
       * @param runnable the {@link Runnable} to be executed.
       */
      public void submitToHead(Runnable runnable) {
          synchronized (queueLock) {
              queue.addFirst(runnable);
  
              maybePollAndExecute();
          }
      }
  
      /**
       * Pause execution of queued {@link java.lang.Runnable}s.
       */
      public void pause() {
          synchronized (queueLock) {
              paused = true;
          }
      }
  
      /**
       * Resume execution of queued {@link java.lang.Runnable}s.
       */
      public void resume() {
          synchronized (queueLock) {
              paused = false;
  
              maybePollAndExecute();
          }
      }
  
      private void maybePollAndExecute() {
          synchronized (queueLock) {
              if (pending < concurrencyLimit && !paused && !queue.isEmpty()) {
                  executor.execute(new Task(queue.poll()));
                  pending++;
              }
          }
      }
  
      private class Task implements Runnable {
  
          private final Runnable runnable;
  
          Task(Runnable runnable) {
              Preconditions.checkNotNull(runnable);
  
              this.runnable = runnable;
          }
  
          @Override
          public void run() {
              try {
                  runnable.run();
              } catch (Throwable throwable) {
                  log.warn("Uncaught Throwable during execution.", throwable);
              }
  
              InlineTask inlineTask = null;
  
              synchronized (queueLock) {
                  if (queue.isEmpty() || paused) {
                      pending--;
                  } else {
                      // pending count remains the same
                      inlineTask = new InlineTask(queue.poll());
                  }
              }
  
              if (inlineTask != null) {
                  inlineTask.run();
              }
          }
      }
  
      private class InlineTask implements Runnable {
  
          private final Runnable runnable;
  
          InlineTask(Runnable runnable) {
              Preconditions.checkNotNull(runnable);
  
              this.runnable = runnable;
          }
  
          @Override
          public void run() {
              try {
                  runnable.run();
              } catch (Throwable throwable) {
                  log.warn("Uncaught Throwable during execution.", throwable);
              }
  
              synchronized (queueLock) {
                  if (queue.isEmpty() || paused) {
                      pending--;
                  } else {
                      // pending count remains the same
                      executor.execute(new Task(queue.poll()));
                  }
              }
          }
      }
  
  }
  