fixed stuff for the new data transfer format

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());
 	}
 	
 	/**

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<DataQueue> queues;
-	
-	/**
-	 * Outgoing queue of this collector
-	 */
 	private EventBus bus;
+	private List<MessageListener> listeners = new ArrayList<>();
+	
+	private List<Metadata> metadata;
+	private boolean first = true;
 	
 	public Collector(EventBus b){
-		queues = new ArrayList<DataQueue>();
 		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<Message> 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<DataQueue> 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<nq;i++){
-			DataQueue q = queues.get(i);
-			for(ComponentMetadata cm: q.getDataMessageMetadata().getComponents()){
-				// The dimension number is the number of queues minus the index of the current queue minus 1 (because dimension index starts at 0)
-				dataMessageMetadata.getComponents().add(new ComponentMetadata(cm.getId(), nq-i-1));
-			}
-		}
-		return(dataMessageMetadata);
-	}
-
 }
+

ch.psi.fda/src/main/java/ch/psi/fda/aq/Manipulator.java

@@ -58,16 +58,20 @@ public class Manipulator {
 					manipulation.initialize(this.metadata);
 					
 					// Add manipulation id to metadata
+					System.out.println(""+manipulation.getId());
 					this.metadata.add(new Metadata(manipulation.getId(),0)); // Calculated component always belongs to lowes dimension
 				}
 			}
 			
 			
 			DataMessage dm = (DataMessage) message;
-			
+//			message = new DataMessage(metadata);
 			for(Manipulation manipulation: manipulations){
+//				((DataMessage)message).getData().add(manipulation.execute(dm));
 				dm.getData().add(manipulation.execute(dm));
 			}
+			// Need to update the metadata of the message
+			dm.setMetadata(this.metadata);
 		}
 		bus.post(message);
 	}

ch.psi.fda/src/main/java/ch/psi/fda/core/ActionLoop.java

@@ -21,7 +21,7 @@ package ch.psi.fda.core;
 
 import java.util.List;
 
-import ch.psi.fda.core.messages.DataQueue;
+import com.google.common.eventbus.EventBus;
 
 /**
  * Loop of actions to accomplish a task. Depending on the loop
@@ -62,5 +62,5 @@ public interface ActionLoop extends Action {
 	 */
 	public void setDataGroup(boolean dataGroup);
 	
-	public DataQueue getDataQueue();
+	public EventBus getEventBus();
 }

ch.psi.fda/src/main/java/ch/psi/fda/core/loops/ActorSensorLoop.java

@@ -21,26 +21,23 @@ package ch.psi.fda.core.loops;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingQueue;
 import java.util.logging.Logger;
 
+import com.google.common.eventbus.EventBus;
+
 import ch.psi.fda.core.Action;
 import ch.psi.fda.core.ActionLoop;
 import ch.psi.fda.core.Actor;
 import ch.psi.fda.core.ActorSetCallable;
 import ch.psi.fda.core.Guard;
 import ch.psi.fda.core.Sensor;
-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.EndOfStreamMessage;
-import ch.psi.fda.core.messages.Message;
+import ch.psi.fda.core.messages.Metadata;
 
 /**
  * Loop of actions to accomplish a task or test.
@@ -98,11 +95,11 @@ public class ActorSensorLoop implements ActionLoop {
 	 */
 	private List<Sensor> 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<Message> 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<Message> 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<ActorSetCallable> pactors;
 	
+	private EventBus eventBus = new EventBus();
+	
 	
 	public ActorSensorLoop(){
 		this(false);
@@ -134,8 +133,6 @@ public class ActorSensorLoop implements ActionLoop {
 		this.actors = new ArrayList<Actor>();
 		this.pactors = new ArrayList<ActorSetCallable>();
 		this.sensors = new ArrayList<Sensor>();
-		
-		this.dataQueue = new LinkedBlockingQueue<Message>(1000);
 	}
 	
 	
@@ -159,6 +156,14 @@ public class ActorSensorLoop implements ActionLoop {
 	public void execute() throws InterruptedException {
 		abort = false;
 		
+		List<Metadata> 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;
 	}
 	
 	

ch.psi.fda/src/main/java/ch/psi/fda/core/messages/ComponentMetadata.java

@@ -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 <http://www.gnu.org/licenses/>.
- * 
- */
-
-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;
-	}
-}

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<Object> data;
-	private final List<Metadata> metadata;
+	private List<Metadata> metadata;
 	
 //	public DataMessage(){
 //		this.data = new ArrayList<Object>();
@@ -48,6 +48,9 @@ public class DataMessage extends Message{
 	public List<Metadata> getMetadata(){
 		return metadata;
 	}
+	public void setMetadata(List<Metadata> metadata){
+		this.metadata = metadata;
+	}
 	
 	// Utility functions
 	

ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataMessageMetadata.java

@@ -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 <http://www.gnu.org/licenses/>.
- * 
- */
-
-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<ComponentMetadata> components;
-	
-	public DataMessageMetadata(){
-		components = new ArrayList<ComponentMetadata>();
-	}
-
-	/**
-	 * @return the list of component metadata of a message described by this object.
-	 */
-	public List<ComponentMetadata> 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<components.size();i++){
-			ComponentMetadata c =components.get(i);
-			if(c.getId().equals(componentId)){
-				return(i);
-			}
-		}
-		throw new IllegalArgumentException("There is no message component with Id "+ componentId);
-	}
-	
-	/**
-	 * Clone data structure
-	 * @return Cloned data structure
-	 */
-	@Override
-	public DataMessageMetadata clone(){
-		DataMessageMetadata metadata = new DataMessageMetadata();
-		
-		for(int i=0;i<components.size();i++){
-			ComponentMetadata c =components.get(i);
-			metadata.getComponents().add(new ComponentMetadata(c.getId(), c.getDimension()));
-		}
-		
-		return(metadata);
-	}
-}

ch.psi.fda/src/main/java/ch/psi/fda/core/messages/DataQueue.java

@@ -1,56 +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 <http://www.gnu.org/licenses/>.
- * 
- */
-
-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<Message> 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<Message> queue, DataMessageMetadata dataMessageMetadata){
-		this.queue = queue;
-		this.dataMessageMetadata = dataMessageMetadata;
-	}
-
-	public BlockingQueue<Message> getQueue() {
-		return queue;
-	}
-
-	public DataMessageMetadata getDataMessageMetadata() {
-		return dataMessageMetadata;
-	}
-}

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;
 	}

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 <http://www.gnu.org/licenses/>.
@@ -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<Message> q1;
-	private BlockingQueue<Message> q2;
-	private BlockingQueue<Message> 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<Message>();
-		q2 = new LinkedBlockingQueue<Message>();
-		q3 = new LinkedBlockingQueue<Message>();
-		
-		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<Metadata> m0;
+		List<Metadata> m1;
+		List<Metadata> 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());
+
 	}
 
 }

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<Message> 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<numberOfSensors; x++){
-			if(! m.getComponents().get(x).getId().equals("id"+x)){
-				fail = true;
-			}
-		}
-		
-		if(fail){
-			fail("Ids of the component metadata elements inside the data message metadata is not correct");
-		}
 	}
 	
 	/**
@@ -223,37 +171,23 @@ public class ActorSensorLoopTest {
 			}
 		});
 		
-		final BlockingQueue<Message> 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();
 	}