From 70089581c4197aa36b34ccf46b5bd5a8e9f12080 Mon Sep 17 00:00:00 2001 From: Simon Ebner Date: Thu, 24 Oct 2013 10:03:23 +0200 Subject: [PATCH] First version with asynchronous execution --- .../ch/psi/fda/rest/AcquisitionEngine.java | 87 ++++++++++++------- .../ch/psi/fda/rest/ExecutionRequest.java | 43 +++++++++ .../ch/psi/fda/rest/services/ScanService.java | 4 +- 3 files changed, 103 insertions(+), 31 deletions(-) create mode 100644 ch.psi.fda/src/main/java/ch/psi/fda/rest/ExecutionRequest.java diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/rest/AcquisitionEngine.java b/ch.psi.fda/src/main/java/ch/psi/fda/rest/AcquisitionEngine.java index 281ea4e..cc0f820 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/rest/AcquisitionEngine.java +++ b/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 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(); } } diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/rest/ExecutionRequest.java b/ch.psi.fda/src/main/java/ch/psi/fda/rest/ExecutionRequest.java new file mode 100644 index 0000000..f25a072 --- /dev/null +++ b/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 . + * + */ +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; + } +} diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/rest/services/ScanService.java b/ch.psi.fda/src/main/java/ch/psi/fda/rest/services/ScanService.java index adf22af..5c6ea1f 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/rest/services/ScanService.java +++ b/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