From 40cc7f6a31abc9d29adf3ce8d37fa172c46c45a3 Mon Sep 17 00:00:00 2001 From: Simon Ebner Date: Mon, 21 Oct 2013 08:41:22 +0200 Subject: [PATCH] fixed stuff for the new data transfer format --- .../main/java/ch/psi/fda/aq/Acquisition.java | 30 +- .../main/java/ch/psi/fda/aq/Collector.java | 136 +++----- .../main/java/ch/psi/fda/aq/Manipulator.java | 6 +- .../main/java/ch/psi/fda/core/ActionLoop.java | 4 +- .../psi/fda/core/loops/ActorSensorLoop.java | 53 ++- .../fda/core/messages/ComponentMetadata.java | 54 --- .../ch/psi/fda/core/messages/DataMessage.java | 5 +- .../core/messages/DataMessageMetadata.java | 79 ----- .../ch/psi/fda/core/messages/DataQueue.java | 56 --- .../ch/psi/fda/core/messages/Metadata.java | 6 +- .../java/ch/psi/fda/aq/CollectorTest.java | 329 ++++++++++-------- .../fda/core/loops/ActorSensorLoopTest.java | 92 +---- 12 files changed, 284 insertions(+), 566 deletions(-) delete mode 100644 ch.psi.fda/src/main/java/ch/psi/fda/core/messages/ComponentMetadata.java delete mode 100644 ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessageMetadata.java delete mode 100644 ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataQueue.java diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/aq/Acquisition.java b/ch.psi.fda/src/main/java/ch/psi/fda/aq/Acquisition.java index e9455ed..03264ce 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/aq/Acquisition.java +++ b/ch.psi.fda/src/main/java/ch/psi/fda/aq/Acquisition.java @@ -24,7 +24,6 @@ import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.ArrayList; -import java.util.Collections; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -140,7 +139,7 @@ public class Acquisition { private Handler logHandler = null; - private Collector col; +// private Collector col; private File datafile; @@ -237,13 +236,12 @@ public class Acquisition { if(smodel.getScan().getManipulation()!= null && smodel.getScan().getManipulation().size()>0){ // Setup optimized with manipulations - EventBus b = new AsyncEventBus(Executors.newCachedThreadPool()); // Map scan to base model // After this call actionLoop and collector will be initialized Collector collector = new Collector(b); mapScan(collector, smodel); - col = collector; +// col = collector; logger.fine("ActionLoop and Collector initialized"); @@ -260,7 +258,7 @@ public class Acquisition { // Setup optimized without manipulations Collector collector = new Collector(bus); mapScan(collector, smodel); - col = collector; +// col = collector; this.serializer = new DataSerializerTXT(datafile, true); bus.register(serializer); @@ -273,24 +271,13 @@ public class Acquisition { */ public void execute() throws InterruptedException { -// acquisitionThread = Thread.currentThread(); - try{ active = true; - Thread tc = new Thread(col); - tc.start(); - actionLoop.prepare(); actionLoop.execute(); actionLoop.cleanup(); - // Wait for data collector threads - // Do this with a Latch or something like that - - // Give the threads 1 minute to catch up - tc.join(60000); - // Send notifications out to all recipients that want to have success notifications try { String hostname = InetAddress.getLocalHost().getHostName(); @@ -417,13 +404,13 @@ public class Acquisition { if(scan.getCdimension() != null){ ActionLoop aLoop = mapContinuousDimension(scan.getCdimension()); actionLoop = aLoop; - collector.getQueues().add(aLoop.getDataQueue()); + collector.addEventBus(aLoop.getEventBus()); } // Map discrete step dimensions for(DiscreteStepDimension d: scan.getDimension()){ ActorSensorLoop l = mapDiscreteStepDimension(d); - collector.getQueues().add(l.getDataQueue()); + collector.addEventBus(l.getEventBus()); if(actionLoop != null){ l.getActionLoops().add(actionLoop); } @@ -442,6 +429,7 @@ public class Acquisition { actionLoop.getPostActions().addAll(mapActions(scan.getPostAction())); + // TODO need to be removed! and done differently !!!! // Handle iterations by adding a pseudo dimension and setting the // datagroup flag in the main loop if(configuration.getNumberOfExecution()>1){ @@ -454,7 +442,7 @@ public class Acquisition { // Set toplevel action loop actionLoop = l; - collector.getQueues().add(l.getDataQueue()); + collector.addEventBus(l.getEventBus()); } // handling manipulations @@ -502,10 +490,6 @@ public class Acquisition { this.manipulations.add(manipulation); } } - - // TODO Remove this workaround - // Revert queues to match sequence - Collections.reverse(collector.getQueues()); } /** diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/aq/Collector.java b/ch.psi.fda/src/main/java/ch/psi/fda/aq/Collector.java index 8a5e43b..1af95d5 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/aq/Collector.java +++ b/ch.psi.fda/src/main/java/ch/psi/fda/aq/Collector.java @@ -21,122 +21,82 @@ package ch.psi.fda.aq; import java.util.ArrayList; import java.util.List; -import java.util.concurrent.BlockingQueue; -import java.util.logging.Logger; import com.google.common.eventbus.EventBus; +import com.google.common.eventbus.Subscribe; -import ch.psi.fda.core.messages.ComponentMetadata; import ch.psi.fda.core.messages.DataMessage; -import ch.psi.fda.core.messages.DataMessageMetadata; -import ch.psi.fda.core.messages.DataQueue; +import ch.psi.fda.core.messages.Metadata; import ch.psi.fda.core.messages.StreamDelimiterMessage; import ch.psi.fda.core.messages.EndOfStreamMessage; import ch.psi.fda.core.messages.Message; /** * Collector class that is collecting and merging data from different Queues. - * @author ebner - * */ -public class Collector implements Runnable{ +public class Collector { - private static Logger logger = Logger.getLogger(Collector.class.getName()); - - /** - * List of queues and their metadata this collector is responsible to readout. - * The logic is as follows: - * Read out the last queue until the first queue in the list has the EndOfLoopMessage. - */ - private List queues; - - /** - * Outgoing queue of this collector - */ private EventBus bus; + private List listeners = new ArrayList<>(); + + private List metadata; + private boolean first = true; public Collector(EventBus b){ - queues = new ArrayList(); this.bus = b; } - @Override - public void run() { - - if(queues.size()>0){ - try{ - readQueue(0, null); - } - catch(InterruptedException e){ - // Readout aborted through interrupt - } - } - - bus.post(new EndOfStreamMessage()); - - - logger.info("END"); - + public void addEventBus(EventBus b){ + MessageListener l = new MessageListener(); + listeners.add(l); + b.register(l); } - private void readQueue(int index, DataMessage m) throws InterruptedException{ + + private class MessageListener{ - BlockingQueue q = queues.get(index).getQueue(); + private DataMessage message; - // Read Message - Message message = q.take(); - while(message instanceof DataMessage){ - // Create new data message - DataMessage dm = new DataMessage(); - if(m!=null){ - dm.getData().addAll(m.getData()); + @Subscribe + public void onMessage(Message message){ + int level = listeners.indexOf(this); + if(message instanceof DataMessage){ + this.message = (DataMessage) message; + + if(level==0){ + if(first){ + metadata = new ArrayList<>(); + for(int i=listeners.size()-1;i>=0;i--){ + // Correct/Add dimension information + for(Metadata m: listeners.get(i).getMessage().getMetadata()){ + m.setDimension(i); + } + metadata.addAll(listeners.get(i).getMessage().getMetadata()); + } + + } + DataMessage m = new DataMessage(metadata); + for(int i=listeners.size()-1;i>=0;i--){ + m.getData().addAll(listeners.get(i).getMessage().getData()); + } + bus.post(m); + } } - - dm.getData().addAll(((DataMessage)message).getData()); - - if(index<(queues.size()-1)){ - readQueue(index+1, dm); + if(message instanceof EndOfStreamMessage){ + + StreamDelimiterMessage ddm = new StreamDelimiterMessage(level, ((EndOfStreamMessage)message).isIflag()); + bus.post(ddm); + if(level==(listeners.size()-1)){ // if highest dimension then send end of stream + bus.post(new EndOfStreamMessage()); + } } - else{ - // Write message to outgoing queue - bus.post(dm); - } - - // Read next message - message = q.take(); } - if(message instanceof EndOfStreamMessage){ - // Translate EndOfStream to StreamDelimiter message - StreamDelimiterMessage ddm = new StreamDelimiterMessage(queues.size()-1-index, ((EndOfStreamMessage)message).isIflag()); - // Write message to outgoing queue - bus.post(ddm); + public DataMessage getMessage(){ + return message; } } - - public List getQueues() { - return queues; - } - - - /** - * Get the outgoing data metadata - */ - public DataMessageMetadata getMetadata(){ - DataMessageMetadata dataMessageMetadata = new DataMessageMetadata(); - - // Generate new combined metadata and add dimension information to the components - int nq = queues.size(); - for(int i=0;i sensors; - /** - * Data queue sensor data is posted to. A message consists of a list of data objects - * that are read out of the sensors of this loop. - */ - private BlockingQueue dataQueue; +// /** +// * Data queue sensor data is posted to. A message consists of a list of data objects +// * that are read out of the sensors of this loop. +// */ +// private BlockingQueue dataQueue; /** * Guard used to check whether the environment was ok while reading out the sensors @@ -116,6 +113,8 @@ public class ActorSensorLoop implements ActionLoop { private List pactors; + private EventBus eventBus = new EventBus(); + public ActorSensorLoop(){ this(false); @@ -134,8 +133,6 @@ public class ActorSensorLoop implements ActionLoop { this.actors = new ArrayList(); this.pactors = new ArrayList(); this.sensors = new ArrayList(); - - this.dataQueue = new LinkedBlockingQueue(1000); } @@ -159,6 +156,14 @@ public class ActorSensorLoop implements ActionLoop { public void execute() throws InterruptedException { abort = false; + List metadata = new ArrayList<>(); + + // Build up data message metadata based on the sensors currently registered. + for(Sensor s: sensors){ + metadata.add(new Metadata(s.getId())); + } + + /** * Thread pool for parallel execution of tasks */ @@ -282,7 +287,7 @@ public class ActorSensorLoop implements ActionLoop { } // Read sensors - DataMessage message = new DataMessage(); + DataMessage message = new DataMessage(metadata); for(Sensor sensor: sensors){ // Readout sensor Object o = sensor.read(); @@ -307,7 +312,7 @@ public class ActorSensorLoop implements ActionLoop { if(guardOK){ // Post a message with the sensor data - dataQueue.put(message); + eventBus.post(message); // Loop all configured ActionLoop objects for(ActionLoop actionLoop: actionLoops){ @@ -327,7 +332,7 @@ public class ActorSensorLoop implements ActionLoop { // Issue end of loop control message // Set iflag of the EndOfStreamMessage according to dataGroup flag of this loop - dataQueue.put(new EndOfStreamMessage(dataGroup)); + eventBus.post(new EndOfStreamMessage(dataGroup)); } if(zigZag){ @@ -399,19 +404,9 @@ public class ActorSensorLoop implements ActionLoop { } - /** - * The structure of the data message depends on the sensors registered at this loop - * at the time this method is called. - * @return the data queue and the metadata of the data messages - */ - public DataQueue getDataQueue() { - DataMessageMetadata m = new DataMessageMetadata(); - - // Build up data message metadata based on the sensors currently registered. - for(Sensor s: sensors){ - m.getComponents().add(new ComponentMetadata(s.getId())); - } - return new DataQueue(dataQueue, m); + @Override + public EventBus getEventBus(){ + return eventBus; } diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/ComponentMetadata.java b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/ComponentMetadata.java deleted file mode 100644 index 2c7779f..0000000 --- a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/ComponentMetadata.java +++ /dev/null @@ -1,54 +0,0 @@ -/** - * - * Copyright 2010 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.core.messages; - -import java.io.Serializable; - -/** - * Metadata of a component of a message. Each component has a global id. - * Optionally the component can also belong to a dimension. However, depending on the - * view the number of the dimension might vary. Therefore the dimension number - * might change during the lifetime of a message (component). - */ -public class ComponentMetadata implements Serializable{ - - private static final long serialVersionUID = 1L; - - private final String id; - private final int dimension; - - public ComponentMetadata(String id){ - this.id = id; - this.dimension = 0; - } - - public ComponentMetadata(String id, int dimension){ - this.id = id; - this.dimension = dimension; - } - - public int getDimension() { - return dimension; - } - - public String getId() { - return id; - } -} diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessage.java b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessage.java index 3cf9f89..bfbb128 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessage.java +++ b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessage.java @@ -30,7 +30,7 @@ public class DataMessage extends Message{ private static final long serialVersionUID = 1L; private final List data; - private final List metadata; + private List metadata; // public DataMessage(){ // this.data = new ArrayList(); @@ -48,6 +48,9 @@ public class DataMessage extends Message{ public List getMetadata(){ return metadata; } + public void setMetadata(List metadata){ + this.metadata = metadata; + } // Utility functions diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessageMetadata.java b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessageMetadata.java deleted file mode 100644 index a21ced7..0000000 --- a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessageMetadata.java +++ /dev/null @@ -1,79 +0,0 @@ -/** - * - * Copyright 2010 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.core.messages; - -import java.io.Serializable; -import java.util.ArrayList; -import java.util.List; - -/** - * Structure to hold the metadata of a component of a message. - */ -private class DataMessageMetadata implements Serializable { - private static final long serialVersionUID = 1L; - - /** - * List of the metadata of the message components - */ - private List components; - - public DataMessageMetadata(){ - components = new ArrayList(); - } - - /** - * @return the list of component metadata of a message described by this object. - */ - public List getComponents() { - return components; - } - - /** - * Get the index of the component with the specified Id - * @param componentId Id of the component to look for - * @return Index of the component - * @throws IllegalArgumentException There is no component with the specified Id - */ - public int getIndex(String componentId) throws IllegalArgumentException { - for(int i=0;i. - * - */ - -package ch.psi.fda.core.messages; - -import java.util.concurrent.BlockingQueue; - - -/** - * Container holding for holding a queue and the metadata of the - * data messages of that queue. - */ -public class DataQueue { - - /** - * Queue serving the messages - */ - private final BlockingQueue queue; - /** - * Metadata of the data messages of the queue - */ - private final DataMessageMetadata dataMessageMetadata; - - /** - * @param queue Data queue - * @param dataMessageMetadata Metadata of the data messages of the queue - */ - public DataQueue(BlockingQueue queue, DataMessageMetadata dataMessageMetadata){ - this.queue = queue; - this.dataMessageMetadata = dataMessageMetadata; - } - - public BlockingQueue getQueue() { - return queue; - } - - public DataMessageMetadata getDataMessageMetadata() { - return dataMessageMetadata; - } -} diff --git a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/Metadata.java b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/Metadata.java index f78ccc9..4d8cf3c 100644 --- a/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/Metadata.java +++ b/ch.psi.fda/src/main/java/ch/psi/fda/core/messages/Metadata.java @@ -32,7 +32,7 @@ public class Metadata implements Serializable{ private static final long serialVersionUID = 1L; private final String id; - private final int dimension; + private int dimension; public Metadata(String id){ this.id = id; @@ -44,6 +44,10 @@ public class Metadata implements Serializable{ this.dimension = dimension; } + public void setDimension(int dimension){ + this.dimension = dimension; + } + public int getDimension() { return dimension; } diff --git a/ch.psi.fda/src/test/java/ch/psi/fda/aq/CollectorTest.java b/ch.psi.fda/src/test/java/ch/psi/fda/aq/CollectorTest.java index 8f05ddd..b19ea50 100644 --- a/ch.psi.fda/src/test/java/ch/psi/fda/aq/CollectorTest.java +++ b/ch.psi.fda/src/test/java/ch/psi/fda/aq/CollectorTest.java @@ -2,15 +2,15 @@ * * Copyright 2010 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 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. + * 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 . @@ -21,8 +21,8 @@ package ch.psi.fda.aq; import static org.junit.Assert.*; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; +import java.util.ArrayList; +import java.util.List; import java.util.logging.Logger; import org.junit.After; @@ -33,131 +33,18 @@ import com.google.common.eventbus.EventBus; import com.google.common.eventbus.Subscribe; import ch.psi.fda.aq.Collector; -import ch.psi.fda.core.messages.ComponentMetadata; import ch.psi.fda.core.messages.ControlMessage; import ch.psi.fda.core.messages.DataMessage; -import ch.psi.fda.core.messages.DataMessageMetadata; -import ch.psi.fda.core.messages.DataQueue; import ch.psi.fda.core.messages.EndOfStreamMessage; import ch.psi.fda.core.messages.Message; +import ch.psi.fda.core.messages.Metadata; public class CollectorTest { - + private static Logger logger = Logger.getLogger(CollectorTest.class.getName()); - private EventBus bus; - - private BlockingQueue q1; - private BlockingQueue q2; - private BlockingQueue q3; - - private DataMessageMetadata m1; - private DataMessageMetadata m2; - private DataMessageMetadata m3; - - + @Before public void setUp() throws Exception { - bus = new EventBus(); - - - // Create blocking queues - q1 = new LinkedBlockingQueue(); - q2 = new LinkedBlockingQueue(); - q3 = new LinkedBlockingQueue(); - - m1 = new DataMessageMetadata(); - m1.getComponents().add(new ComponentMetadata("id2-0")); - m2 = new DataMessageMetadata(); - m2.getComponents().add(new ComponentMetadata("id1-0")); - m3 = new DataMessageMetadata(); - m3.getComponents().add(new ComponentMetadata("id0-0")); - - // Dimension y 1,2,3 - DataMessage m = new DataMessage(); - m.getData().add(1d); - q1.put(m); - m = new DataMessage(); - m.getData().add(2d); - q1.put(m); - m = new DataMessage(); - m.getData().add(3d); - q1.put(m); - q1.put(new EndOfStreamMessage()); - - // Dimension x 0.1,0.2 - m = new DataMessage(); - m.getData().add(0.1); - q2.put(m); - m = new DataMessage(); - m.getData().add(0.2); - q2.put(m); - q2.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.1); - q2.put(m); - m = new DataMessage(); - m.getData().add(0.2); - q2.put(m); - q2.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.1); - q2.put(m); - m = new DataMessage(); - m.getData().add(0.2); - q2.put(m); - q2.put(new EndOfStreamMessage()); - - // Dimension - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - - m = new DataMessage(); - m.getData().add(0.01); - q3.put(m); - m = new DataMessage(); - m.getData().add(0.02); - q3.put(m); - q3.put(new EndOfStreamMessage()); - } @After @@ -166,44 +53,180 @@ public class CollectorTest { /** * Test method for {@link ch.psi.fda.aq.Collector#run()}. - * @throws InterruptedException + * + * @throws InterruptedException */ @Test public void testRun() throws InterruptedException { + + EventBus bus = new EventBus(); + + // Metadata + List m0; + List m1; + List m2; + m0 = new ArrayList<>(); + m0.add(new Metadata("id0-0", 0)); + m1 = new ArrayList<>(); + m1.add(new Metadata("id1-0", 1)); + m2 = new ArrayList<>(); + m2.add(new Metadata("id2-0", 2)); + + EventBus b0 = new EventBus(); + EventBus b1 = new EventBus(); + EventBus b2 = new EventBus(); + Collector collector = new Collector(bus); - collector.getQueues().add(new DataQueue(q1, m1)); - collector.getQueues().add(new DataQueue(q2, m2)); - collector.getQueues().add(new DataQueue(q3, m3)); - + collector.addEventBus(b0); + collector.addEventBus(b1); + collector.addEventBus(b2); + // Check component metadata of output queue - int c=2; - for(ComponentMetadata cm: collector.getMetadata().getComponents()){ - logger.info(cm.toString()); - if(cm.getDimension() != c){ - fail("Dimension number does not match required dimension number"); - } - if(! cm.getId().equals("id"+c+"-0")){ - fail("Id does not match required id"); - } - c--; - } - - + // check wether messages arrive - bus.register(new Object(){ + bus.register(new Object() { + + private boolean first = true; + @Subscribe - public void onMessage(Message m){ - if(m instanceof DataMessage){ + public void onMessage(Message m) { + if (m instanceof DataMessage) { DataMessage x = (DataMessage) m; - logger.info( x.toString() ); - } - else if(m instanceof ControlMessage){ - logger.info("---- "+m.toString()+" ----"); + + if (first) { + first = false; + // Check metadata + int c = 2; + for (Metadata cm : x.getMetadata()) { + logger.info(String.format("id: %s dimension: %s", cm.getId(),cm.getDimension())); + if (cm.getDimension() != c) { + fail("Dimension number does not match required dimension number"); + } + if (!cm.getId().equals("id" + c + "-0")) { + fail("Id does not match required id"); + } + c--; + } + } + + logger.info(x.toString()); + } else if (m instanceof ControlMessage) { + logger.info("---- " + m.toString() + " ----"); } } }); - - collector.run(); + + // dimension 2 + DataMessage m = new DataMessage(m2); + m.getData().add(1.0); + b2.post(m); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.1); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.2); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + b1.post(new EndOfStreamMessage()); + + // dimension 2 + m = new DataMessage(m2); + m.getData().add(2.0); + b2.post(m); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.1); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.2); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + b1.post(new EndOfStreamMessage()); + + // dimension 2 + m = new DataMessage(m2); + m.getData().add(3.0); + b2.post(m); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.1); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + m = new DataMessage(m1); + m.getData().add(0.2); + b1.post(m); + + // dimension 0 + m = new DataMessage(m0); + m.getData().add(0.01); + b0.post(m); + m = new DataMessage(m0); + m.getData().add(0.02); + b0.post(m); + b0.post(new EndOfStreamMessage()); + + // dimension 1 + b1.post(new EndOfStreamMessage()); + + // dimension 2 + b2.post(new EndOfStreamMessage()); + } } diff --git a/ch.psi.fda/src/test/java/ch/psi/fda/core/loops/ActorSensorLoopTest.java b/ch.psi.fda/src/test/java/ch/psi/fda/core/loops/ActorSensorLoopTest.java index 735c82c..381d285 100644 --- a/ch.psi.fda/src/test/java/ch/psi/fda/core/loops/ActorSensorLoopTest.java +++ b/ch.psi.fda/src/test/java/ch/psi/fda/core/loops/ActorSensorLoopTest.java @@ -24,7 +24,6 @@ import static org.junit.Assert.*; import java.util.ArrayList; import java.util.HashMap; import java.util.List; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeoutException; import java.util.logging.Level; @@ -34,6 +33,8 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import com.google.common.eventbus.Subscribe; + import ch.psi.fda.TestChannels; import ch.psi.fda.core.Actor; import ch.psi.fda.core.Guard; @@ -43,8 +44,6 @@ import ch.psi.fda.core.actors.ChannelAccessLinearActuator; import ch.psi.fda.core.guard.ChannelAccessGuard; import ch.psi.fda.core.guard.ChannelAccessGuardCondition; import ch.psi.fda.core.loops.ActorSensorLoop; -import ch.psi.fda.core.messages.DataMessage; -import ch.psi.fda.core.messages.DataMessageMetadata; import ch.psi.fda.core.messages.Message; import ch.psi.fda.core.sensors.ChannelAccessSensor; import ch.psi.jcae.Channel; @@ -104,66 +103,15 @@ public class ActorSensorLoopTest { @Test public void testExecute() throws InterruptedException { - final BlockingQueue dataQueue = loopOne.getDataQueue().getQueue(); - Thread t = new Thread(new Runnable() { - - @Override - public void run() { - try { - while(true){ - Message m = dataQueue.take(); - if(m instanceof DataMessage){ - DataMessage x = (DataMessage) m; - logger.fine( x.toString() ); - } - } - } catch (InterruptedException e) { - logger.log(Level.SEVERE, "An Exception occured while reading data from the data queue", e); - // TODO Auto-generated catch block - e.printStackTrace(); - } + loopOne.getEventBus().register(new Object(){ + @Subscribe + public void onMessage(Message m){ + logger.info( m.toString() ); } + }); - t.start(); loopOne.execute(); -// Thread.sleep(3000); -// t.interrupt(); - } - - /** - * Test method for {@link ch.psi.fda.core.loops.ActorSensorLoop#getDataMessageMetadata()}. - * @throws TimeoutException - * @throws InterruptedException - * @throws ChannelException - */ - @Test - public void testGetDataMessageMetadata() throws ChannelException, InterruptedException, TimeoutException{ - ActorSensorLoop loop = new ActorSensorLoop(); - - int numberOfSensors = 2; - Sensor s1 = new ChannelAccessSensor<>("id0", cservice.createChannel(new ChannelDescriptor<>(Double.class, boChannel))); - Sensor s2 = new ChannelAccessSensor<>("id1", cservice.createChannel(new ChannelDescriptor<>(Double.class, aoChannel))); - loop.getSensors().add(s1); - loop.getSensors().add(s2); - - DataMessageMetadata m = loop.getDataQueue().getDataMessageMetadata(); - - - if(m.getComponents().size() != numberOfSensors){ - fail("Loop returned wrong number of components inside the data message metadata"); - } - - boolean fail = false; - for(int x=0;x dataQueue = loop.getDataQueue().getQueue(); - Thread t = new Thread(new Runnable() { - - @Override - public void run() { - try { - while(true){ - Message m = dataQueue.take(); - if(m instanceof DataMessage){ - DataMessage x = (DataMessage) m; - logger.fine( x.toString() ); - } - } - } catch (InterruptedException e) { - logger.log(Level.SEVERE, "An Exception occured while reading data from the data queue", e); - } + + loop.getEventBus().register(new Object(){ + @Subscribe + public void onMessage(Message m){ + logger.info( m.toString() ); } + }); // Set the guard channel to not OK b.setValue(okValue+1); - t.start(); tguard.start(); loop.prepare(); loop.execute(); logger.info("End of test"); - -// Thread.sleep(3000); -// t.interrupt(); }