Add support for combined simulation to MATSim and OMNeT simulations

jdeeco-matsim-plugin/src/cz/cuni/mff/d3s/jdeeco/matsim/plugin/MATSimSimulation.java

@@ -1,6 +1,5 @@
 package cz.cuni.mff.d3s.jdeeco.matsim.plugin;
 
-import java.io.File;
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -10,6 +9,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.TreeSet;
+import java.util.concurrent.Exchanger;
 
 import org.matsim.api.core.v01.Id;
 import org.matsim.api.core.v01.TransportMode;
@@ -60,25 +60,26 @@ public class MATSimSimulation implements DEECoPlugin {
 	private long currentMilliseconds;
 	private final long simulationStep; // in milliseconds
 	private final TravelTime travelTime;
-	private final Controler controler;
+	protected final Controler controler;
 	private final JDEECoWithinDayMobsimListener listener;
 	private final MATSimDataProvider matSimProvider;
 	private final MATSimDataReceiver matSimReceiver;
 	private final Map<Integer, MATSimHost> hosts = new HashMap<>();
 	private final MATSimExtractor extractor;
-	// private final Exchanger<Object> exchanger = new Exchanger<Object>();
+	protected final Exchanger<Object> exchanger;
 
 	public MATSimSimulation(String configPath, AdditionAwareAgentSource... additionalAgentSources) throws IOException {
-		this(new File(configPath), additionalAgentSources);
+		this(configPath, null, additionalAgentSources);
 	}
 
-	public MATSimSimulation(File config, AdditionAwareAgentSource... additionalAgentSources) throws IOException {
+	public MATSimSimulation(String configPath, Exchanger<Object> exchanger, AdditionAwareAgentSource... additionalAgentSources) throws IOException {
+		this.exchanger = exchanger;
 		List<AdditionAwareAgentSource> agentSources = new LinkedList<>();
 		agentSources.add(agentSource);
 		agentSources.addAll(Arrays.asList(additionalAgentSources));
 
 		// Setup MATSim controller
-		controler = new MATSimPreloadingControler(config.getAbsolutePath());
+		controler = new MATSimPreloadingControler(configPath);
 		controler.setOverwriteFiles(true);
 		controler.getConfig().getQSimConfigGroup().setSimStarttimeInterpretation("onlyUseStarttime");
 
@@ -88,10 +89,14 @@ public MATSimSimulation(File config, AdditionAwareAgentSource... additionalAgent
 		final double step = this.controler.getConfig().getQSimConfigGroup().getTimeStepSize();
 		Log.i("Starting simulation: matsimStartTime: " + start + " matsimEndTime: " + end);
 
-		this.extractor = new DefaultMATSimExtractor();
-		this.listener = new JDEECoWithinDayMobsimListener(timer, new DefaultMATSimUpdater(), extractor);
-		this.matSimProvider = (MATSimDataProvider) matSimProviderReceiver;
-		this.matSimReceiver = (MATSimDataReceiver) matSimProviderReceiver;
+		extractor = new DefaultMATSimExtractor();
+		if(exchanger == null) {
+			listener = new JDEECoWithinDayMobsimListener(timer, new DefaultMATSimUpdater(), extractor);
+		} else {
+			listener = new JDEECoWithinDayMobsimListener(exchanger, new DefaultMATSimUpdater(), extractor);
+		}
+		matSimProvider = (MATSimDataProvider) matSimProviderReceiver;
+		matSimReceiver = (MATSimDataReceiver) matSimProviderReceiver;
 
 		Set<String> analyzedModes = new HashSet<String>();
 		analyzedModes.add(TransportMode.car);
@@ -121,7 +126,7 @@ public void notifyStartup(StartupEvent event) {
 		router = new MATSimRouter(controler, travelTime, 10 /* TODO: FAKE VALUE */);
 	}
 
-	public SimulationTimer getTimer() {
+	public Timer getTimer() {
 		return timer;
 	}
 

jdeeco-matsim-plugin/src/cz/cuni/mff/d3s/jdeeco/matsim/simulation/DefaultMATSimUpdater.java

@@ -12,8 +12,8 @@ public void updateJDEECoAgents(Object input, Collection<JDEECoAgent> agents) {
 			Map<String, MATSimInput> map = (Map<String, MATSimInput>) input;
 			for (JDEECoAgent agent: agents) {
 					if (map.containsKey(agent.getId())) {
-						agent.setRoute(((MATSimInput) map.get(agent.getId())).route);
-						agent.setSpeed(((MATSimInput) map.get(agent.getId())).speed);
+						agent.setRoute(map.get(agent.getId()).route);
+						agent.setSpeed(map.get(agent.getId()).speed);
 					}
 			}
 		}

jdeeco-matsim-plugin/src/cz/cuni/mff/d3s/jdeeco/matsim/simulation/JDEECoWithinDayMobsimListener.java

@@ -72,13 +72,15 @@ public Collection<JDEECoAgent> getAllJDEECoAgents() {
 
 	@SuppressWarnings("rawtypes")
 	public void notifyMobsimBeforeSimStep(MobsimBeforeSimStepEvent event) {
-		if (this.remainingExchanges == 0) {
+	/*	if (this.remainingExchanges == 0) {
 			return;
-		}
+		}*/
 		if (exchanger != null) {
 			try {
 				//System.out.println("MATSIM before: " + event.getSimulationTime());
-				updateJDEECoAgents(exchanger.exchange(extractor.extractFromMATSim(getAllJDEECoAgents(), event.getQueueSimulation())));
+				Object out = extractor.extractFromMATSim(getAllJDEECoAgents(), event.getQueueSimulation());
+				Object in = exchanger.exchange(out);
+				updateJDEECoAgents(in);
 				//System.out.println("MATSIM after: " + event.getSimulationTime());
 			} catch (Exception e) {
 				Log.e("jDEECoWithinDayMobsimListener: ", e);

jdeeco-omnet-plugin/src/cz/cuni/mff/d3s/jdeeco/network/omnet/OMNeTSimulation.java

@@ -3,9 +3,11 @@
 import java.io.File;
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 import cz.cuni.mff.d3s.deeco.runtime.DEECoContainer;
 import cz.cuni.mff.d3s.deeco.runtime.DEECoPlugin;
@@ -16,7 +18,24 @@
 import cz.cuni.mff.d3s.jdeeco.network.address.IPAddress;
 
 public class OMNeTSimulation implements DEECoPlugin {
+	public interface BindedSimulation {
+		void run(long duration);
+	}
+
 	public class Timer implements SimulationTimer {
+		private long duration;
+		private Set<BindedSimulation> bindedSimulations = new HashSet<>();
+
+		/**
+		 * Adds binded simulation to be executed together with OMNeT
+		 * 
+		 * @param simulation
+		 *            Simulation to be added
+		 */
+		public void addBindedSimulation(BindedSimulation simulation) {
+			bindedSimulations.add(simulation);
+		}
+
 		@Override
 		public void notifyAt(long time, TimerEventListener listener, DEECoContainer container) {
 			hosts.get(container.getId()).setEventListener(time, listener);
@@ -33,48 +52,60 @@ public long getCurrentMilliseconds() {
 
 		@Override
 		public void start(long duration) {
+			this.duration = duration;
+
+			// Run all registered binded simulations
+			for (BindedSimulation runner : bindedSimulations) {
+				runner.run(duration);
+			}
+
 			try {
 				File config = OMNeTSimulation.this.getOmnetConfig(duration);
 				OMNeTNative.nativeRun("Cmdenv", config.getAbsolutePath());
 			} catch (IOException e) {
 				System.err.println("Failed to start simulation: " + e.getMessage());
 			}
 		}
+
+		public long getDuration() {
+			return duration;
+		}
 	}
 
 	class OMNeTHost implements OMNeTNativeListener {
 		public final DEECoContainer container;
-		
+
 		TimerEventListener eventListener = null;
 		OMNeTBroadcastDevice broadcastDevice = null;
 		OMNeTInfrastructureDevice infrastructureDevice = null;
-		
+
 		OMNeTHost(DEECoContainer container) {
 			this.container = container;
 		}
 
 		public int getId() {
 			return container.getId();
 		}
-		
+
 		public void setEventListener(long time, TimerEventListener listener) {
 			// Register listener and schedule event
 			eventListener = listener;
 			OMNeTNative.nativeCallAt(OMNeTNative.timeToOmnet(time), getId());
 		}
-		
+
 		public void setBroadcastDevice(OMNeTBroadcastDevice device) {
 			broadcastDevice = device;
 		}
-		
+
 		public void setInfrastructureDevice(OMNeTInfrastructureDevice device) {
 			infrastructureDevice = device;
 		}
-		
+
 		public void sendInfrastructurePacket(byte[] packet, IPAddress address) {
-			OMNeTNative.nativeSendPacket(getId(), packet, address.ipAddress/*"MANET.node[" + address.ipAddress + "]"*/);
+			OMNeTNative
+					.nativeSendPacket(getId(), packet, address.ipAddress/* "MANET.node[" + address.ipAddress + "]" */);
 		}
-		
+
 		public void sendBroadcastPacket(byte[] packet) {
 			OMNeTNative.nativeSendPacket(getId(), packet, "");
 		}
@@ -86,33 +117,33 @@ public void at(double absoluteTime) {
 
 		@Override
 		public void packetReceived(byte[] packet, double rssi) {
-			if(rssi == -1 && infrastructureDevice != null) {
+			if (rssi == -1 && infrastructureDevice != null) {
 				infrastructureDevice.receivePacket(packet);
 			}
-			
-			if(rssi >= 0 && broadcastDevice != null) {
+
+			if (rssi >= 0 && broadcastDevice != null) {
 				broadcastDevice.receivePacket(packet, rssi);
 			}
 		}
 	}
-	
+
 	private final Map<Integer, OMNeTHost> hosts = new HashMap<Integer, OMNeTSimulation.OMNeTHost>();
 	private Timer timeProvider = new Timer();
-	
+
 	public File getOmnetConfig(long limit) throws IOException {
 		OMNeTConfigGenerator generator = new OMNeTConfigGenerator(limit);
-		
-		for(OMNeTHost host: hosts.values()) {
+
+		for (OMNeTHost host : hosts.values()) {
 			generator.addNode(host);
 		}
-		
+
 		return generator.writeToOmnet();
 	}
 
-	public SimulationTimer getTimer() {
+	public Timer getTimer() {
 		return timeProvider;
 	}
-	
+
 	public OMNeTHost getHost(int id) {
 		return hosts.get(id);
 	}