First version with asynchronous execution

ch.psi.fda/src/main/java/ch/psi/fda/rest/AcquisitionEngine.java

@@ -18,6 +18,11 @@
  */
 package ch.psi.fda.rest;
 
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.logging.Logger;
 
 import javax.inject.Inject;
@@ -30,8 +35,7 @@ import ch.psi.fda.model.v1.Configuration;
 import ch.psi.jcae.ChannelService;
 
 /**
- * @author ebner
- *
+ * Main engine for data acquisition.
  */
 public class AcquisitionEngine {
 	
@@ -42,45 +46,70 @@ public class AcquisitionEngine {
 	@Inject
 	private ChannelService cservice;
 	
-	private Acquisition acquisition;
+	/**
+	 * Variable holding the current acquisition
+	 */
+	private volatile Acquisition acquisition;
 	
-	public AcquisitionEngine(){
-	}
+	private ExecutorService eservice;
+	
+	private BlockingQueue<ExecutionRequest> requests = new LinkedBlockingQueue<>();
 	
 	@Inject
-	public AcquisitionEngine(ChannelService cservice, AcquisitionConfiguration config){
-		this.cservice=cservice;
+	public AcquisitionEngine(ChannelService cservice, AcquisitionConfiguration config) {
+		this.cservice = cservice;
 		this.config = config;
+
+		// Start main execution loop 
+		eservice = Executors.newSingleThreadExecutor();
+		eservice.execute(new Runnable() {
+			@Override
+			public void run() {
+				try {
+					while (!Thread.currentThread().isInterrupted()) {
+						try {
+							ExecutionRequest r = requests.take();
+
+							logger.info("Execute - " + r.getTrackingId());
+
+							EventBus ebus = new EventBus();
+
+							synchronized (AcquisitionEngine.this) { // synchronize access to acquisition object via the AcquisitionEngine object
+								acquisition = new Acquisition(AcquisitionEngine.this.cservice, AcquisitionEngine.this.config);
+							}
+							acquisition.initalize(ebus, r.getConfiguration());
+							acquisition.execute();
+							logger.info("" + r.getTrackingId() + " done");
+						} finally {
+							acquisition.destroy();
+							acquisition = null;
+						}
+					}
+				} catch (InterruptedException e) {
+				}
+			}
+		});
 	}
 	
-	public void scan(Configuration configuration){
-		if(acquisition!=null){
-			throw new RuntimeException("There is already and acquisition running");
-		}
-		
+	
+	public String submit(Configuration configuration){
+		ExecutionRequest r = new ExecutionRequest(UUID.randomUUID().toString(), configuration);
 		try{
-			EventBus ebus = new EventBus();
-			acquisition = new Acquisition(cservice, config);
-			acquisition.initalize(ebus, configuration);
-			acquisition.execute();
-		}
-		catch(InterruptedException e){
-			throw new RuntimeException(e);
-		}
-		finally{
-			acquisition.destroy();
-			acquisition=null;
+			requests.put(r);
+		} catch (InterruptedException e) {
 		}
+		return r.getTrackingId();
 	}
 	
 	
 	public void stop(){
-		if(acquisition==null){
-			return;
+		synchronized(this){
+			if(acquisition==null){
+				return;
+			}
+			
+			logger.info("Stop current acquisition");
+			acquisition.abort();
 		}
-		
-		logger.info("Stop acquisition");
-		acquisition.abort();
-//		acquisition.destroy();
 	}
 }

ch.psi.fda/src/main/java/ch/psi/fda/rest/ExecutionRequest.java

@@ -0,0 +1,43 @@
+/**
+ * 
+ * Copyright 2013 Paul Scherrer Institute. All rights reserved.
+ * 
+ * This code is free software: you can redistribute it and/or modify it under
+ * the terms of the GNU Lesser General Public License as published by the Free
+ * Software Foundation, either version 3 of the License, or (at your option) any
+ * later version.
+ * 
+ * This code is distributed in the hope that it will be useful, but without any
+ * warranty; without even the implied warranty of merchantability or fitness for
+ * a particular purpose. See the GNU Lesser General Public License for more
+ * details.
+ * 
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this code. If not, see <http://www.gnu.org/licenses/>.
+ * 
+ */
+package ch.psi.fda.rest;
+
+import ch.psi.fda.model.v1.Configuration;
+
+/**
+ * @author ebner
+ *
+ */
+public class ExecutionRequest {
+	
+	private final String trackingId;
+	private final Configuration configuration;
+	
+	public ExecutionRequest(String trackingId, Configuration configuration){
+		this.trackingId = trackingId;
+		this.configuration = configuration;
+	}
+
+	public String getTrackingId() {
+		return trackingId;
+	}
+	public Configuration getConfiguration() {
+		return configuration;
+	}
+}

ch.psi.fda/src/main/java/ch/psi/fda/rest/services/ScanService.java

@@ -38,8 +38,8 @@ public class ScanService {
 	@POST
 	@Consumes({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
 //	@Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
-	public void execute(Configuration configuration) throws InterruptedException{
-		aengine.scan(configuration);
+	public String execute(Configuration configuration) throws InterruptedException{
+		return aengine.submit(configuration);
 	}
 	
 	@DELETE