/**
 * @license
 * Copyright 2019 The FOAM Authors. All Rights Reserved.
 * http://www.apache.org/licenses/LICENSE-2.0
 */

package foam.util.concurrent;

import foam.core.Agency;
import foam.core.ContextAgent;
import foam.core.X;
import java.util.concurrent.Semaphore;

/**
* An Aynchronous implementation of the AssemblyLine interface.
* Uses the threadpool to avoid blocking the caller.
**/
public class AsyncAssemblyLine
  extends SyncAssemblyLine
{
  protected Agency pool_;
  protected String agencyName_ = "AsyncAssemblyLine";

  public AsyncAssemblyLine(X x) {
    this(x, "threadPool");
  }

  public AsyncAssemblyLine(X x, String agencyName) {
    super(x);
    pool_ = (Agency) x.get("threadPool");
    agencyName_ += ":" + agencyName;
  }

  public void enqueue(Assembly job) {
    if ( shutdown_ ) throw new IllegalStateException("Can't enqueue into a shutdown AssemblyLine.");

    final Assembly[] previous = new Assembly[1];

    synchronized ( startLock_ ) {
      synchronized ( qLock_ ) {
        previous[0] = q_;
        q_ = job;
      }
      try {
        job.executeUnderLock();
        job.startJob();
      } catch (Throwable t) {
        synchronized ( qLock_ ) {
          q_ = previous[0];
        }
        throw t;
      }
    }

    pool_.submit(x_, new ContextAgent() { public void execute(X x) {
      try {
        job.executeJob();

        if ( previous[0] != null ) previous[0].waitToComplete();
        previous[0] = null;

        boolean isLast = false;
        synchronized ( qLock_ ) {
          // If I'm still the only job in the queue, then remove me
          if ( q_ == job ) {
            q_ = null;
            isLast = true;
          }
        }

        synchronized ( endLock_ ) {
          try {
            job.endJob(isLast);
          } catch (Throwable t) {
            ((foam.nanos.logger.Logger) x.get("logger")).error(this.getClass().getSimpleName(), agencyName_, t);
          }
        }
      } finally {
        job.complete();
      }
    }}, agencyName_);
  }

  public void shutdown() {
    try {
      Semaphore s = new Semaphore(1);
      s.acquire();

      enqueue(new AbstractAssembly() {
        public void startJob() {
          shutdown_ = true;
        }
        public void endJob() {
          s.release();
        }
      });
      s.acquire();
    } catch (InterruptedException e) {
    }
  }
}

@@ -779,24 +779,110 @@
 mbly job) %7B%0A
+    final Assembly%5B%5D previous = new Assembly%5B1%5D;%0A%0A    synchronized ( startLock_ ) %7B%0A  
     if ( shu
@@ -975,92 +975,8 @@
 );%0A%0A
-    final Assembly%5B%5D previous = new Assembly%5B1%5D;%0A%0A    synchronized ( startLock_ ) %7B%0A
     
@@ -2253,24 +2253,38 @@
 void endJob(
+boolean isLast
 ) %7B%0A        
@@ -2372,24 +2372,113 @@
 eption e) %7B%0A
+    %7D catch (IllegalStateException e) %7B%0A      // Line is already shutdown, so no problem%0A
     %7D%0A  %7D%0A%7D%0A

Fix AsyncAssemblinLine shutdown.