Adds first Byte of MQTT packets, Improves MQTT protocol and Protocols

Changes SimpleImplementations to use Ports

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/Packet.java

@@ -12,12 +12,23 @@ public abstract class Packet {
 	 */
 	protected long timestamp;
 	
+	/**
+	 * SourcePort of the packet
+	 */
+	protected Port source;
+	
+	/**
+	 * DestinationPort of the packet
+	 */
+	protected Port destination;
 	/**
 	 * Creates a new packet with the given timestamp
 	 * @param timestamp time the packet was sent
 	 */
-	protected Packet(long timestamp){
+	protected Packet(long timestamp, Port source, Port destination){
 		this.timestamp = timestamp;
+		this.source = source;
+		this.destination = destination;
 	}
 	/**
 	 * 
@@ -47,4 +58,31 @@ public abstract class Packet {
 	public long getTimestamp(){
 		return timestamp;
 	}
+	
+	/**
+	 * Sets the timestamp to the given value
+	 * 
+	 * @param timestamp timstamp to set
+	 */
+	public void setTimestamp(long timestamp){
+		this.timestamp = timestamp;
+	}
+	
+	/**
+	 * Returns the source port of the packet
+	 * 
+	 * @return source port
+	 */
+	public Port getSource(){
+		return source;
+	}
+	
+	/**
+	 * Returns the destination port of the packet
+	 * 
+	 * @return destination port
+	 */
+	public Port getDestination(){
+		return destination;
+	}
 }

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/Protocol.java

@@ -19,7 +19,7 @@ public interface Protocol {
 	 * @param packetLost True if the packet was lost
 	 * @return next Packet, which was sent
 	 */
-	public Packet generateNextPaket(Port port, long timestep, boolean packetLost);
+	public Collection<Packet> generateNextPakets(Port port, long timestep, boolean packetLost);
 
 	/**
 	 * Returns the number of different roles the participating SmartDevice(Port)s

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/protocols/MQTT_protocol.java

@@ -21,22 +21,98 @@ public class MQTT_protocol implements Protocol {
 	 * Broker which collects and distributes messages
 	 */
 	private Port broker;
+
 	/**
 	 * Publishers like sensors, which publish data
 	 */
 	private LinkedList<Port> pubs;
+
 	/**
 	 * Subscriber which subscribe to different Topics
 	 */
 	private LinkedList<Port> subs;
+
 	/**
-	 * Devices that are Publisher and Subscriber and therefore send and receive messages
+	 * Devices that are Publisher and Subscriber and therefore send and receive
+	 * messages
 	 */
 	private LinkedList<Port> pubSubs;
+
+	/**
+	 * Packets which are currently being generated or were generated after the
+	 * last generatePackets Call
+	 */
+	private LinkedList<Packet> currentPackets = new LinkedList<Packet>();
+	
+	/**
+	 * Topics of the MQTT Broker
+	 */
+	private LinkedList<String> topics = new LinkedList<String>();
+
+	/**
+	 * Creates a new MQTT Protocol
+	 * 
+	 * @param broker broker of the protocol
+	 */
+	public MQTT_protocol(Port broker) {
+		this.broker = broker;
+	    topics.add("temperature");
+	    topics.add("doorState");
+	}
+	
 	@Override
-	public Packet generateNextPaket(Port port, long timestep, boolean packetLost) {
+	public Collection<Packet> generateNextPakets(Port port, long timestep, boolean packetLost) {
+		/**
+		 * Packets which will be generated
+		 */
+		LinkedList<Packet> returnPackets = new LinkedList<Packet>();
+		//Update all Timestamps of previous created Packets
+		for(Packet p: currentPackets){
+			p.setTimestamp(p.getTimestamp()+timestep);
+		}
+		//Add these packets to the return
+		returnPackets.addAll(currentPackets);
+		//remove packets from the old list
+		currentPackets.clear();
+		
+		/**
+		 * Update the lastTime the port was triggered
+		 */
 		port.setLastTrigger(timestep);
-		return new MQTT_packet(timestep);
+		
+		/**
+		 * Generate new Packets regarding to their class
+		 */
+		if(port==broker){
+			//Broker should not send new packages, without new messages
+		}else if(pubs.contains(port)){
+			/**
+			 * Message to be published (Should be further specialized by a Sensor Device) 
+			 */
+			String msg = "Topic: DoorState:"+(Math.random()<0.5?"true":"false");
+			returnPackets.add(new MQTT_packet(MQTT_packet.PUBLISH, timestep, port, broker, msg));
+			//Response
+			timestep+=broker.getResponseTime();
+			returnPackets.add(new MQTT_packet(MQTT_packet.PUBACK, timestep, broker, port));
+			
+			//Publish to Subscribers
+			//Should be be improved to just notify Subs that are subscribed to the topic
+			for(Port p:subs){
+				timestep+=broker.getResponseTime();
+				returnPackets.add(new MQTT_packet(MQTT_packet.PUBLISH, timestep, broker, p, msg));
+				returnPackets.add(new MQTT_packet(MQTT_packet.PUBACK, timestep+p.getResponseTime(), p, broker));
+			}
+			for(Port p:pubSubs){
+				timestep+=broker.getResponseTime();
+				returnPackets.add(new MQTT_packet(MQTT_packet.PUBLISH, timestep, broker, p, msg));
+				returnPackets.add(new MQTT_packet(MQTT_packet.PUBACK, timestep+p.getResponseTime(), p, broker));
+			}
+		}else if(subs.contains(port)){
+			returnPackets.add(new MQTT_packet(MQTT_packet.SUBSCRIBE, timestep, port, broker, "Topic: DoorState"));
+			timestep+=broker.getResponseTime();
+			returnPackets.add(new MQTT_packet(MQTT_packet.SUBACK, timestep, broker, port));
+		}
+		return null;
 	}
 
 	@Override
@@ -46,8 +122,8 @@ public class MQTT_protocol implements Protocol {
 
 	@Override
 	public String[] getRoles() {
-		//PublisherSubscriber is Publisher as well as Subscriber
-		return new String[]{"Broker", "PublisherSubsriber","Publisher","Subscriber"};
+		// PublisherSubscriber is Publisher as well as Subscriber
+		return new String[] { "Broker", "PublisherSubsriber", "Publisher", "Subscriber" };
 	}
 
 	@Override
@@ -69,33 +145,60 @@ public class MQTT_protocol implements Protocol {
 
 	@Override
 	public boolean addDeviceOfRole(Port device, int role) {
-		switch (role) {
-		case 0:
-			if(broker != null)
-				return false;
-			else{
+		/*
+		 * First device has to be the Broker 
+		 */
+		if (broker==null) {
+			if (role == 0) {
 				broker = device;
+				return true;
+			} else {
+				return false;
 			}
-			break;
+		}
+		switch (role) {
+		case 0:
+			//Just one broker allowed.
+			return false;
 		case 1:
 			pubSubs.add(device);
+			break;
 		case 2:
 			pubs.add(device);
+			break;
 		case 3:
 			subs.add(device);
+			break;
 		default:
+			// invalid role
 			return false;
 		}
+		// Create packets for the connecting Client
+		currentPackets.add(new MQTT_packet(MQTT_packet.CONNECT, currentPackets.size()/2, device, broker));
+		currentPackets.add(new MQTT_packet(MQTT_packet.CONNACK, currentPackets.size()/2+1, broker, device));
 		return true;
 	}
 
 	@Override
 	public void removeDevice(Port device) {
-		if(broker == device)
+		/**
+		 * true if the device was removed
+		 */
+		boolean removedDevice=false;
+		if (broker == device){
 			broker = null;
-		pubSubs.remove(device);
-		subs.remove(device);
-		pubs.remove(device);
+		}
+		removedDevice|=pubSubs.remove(device);
+		removedDevice|=subs.remove(device);
+		removedDevice|=pubs.remove(device);
+		if(removedDevice && broker!=null){
+			//If not the broker and device was removed -> disconnect
+			currentPackets.add(new MQTT_packet(MQTT_packet.DISCONNECT, currentPackets.size()/2, device, broker));
+		}
+	}
+	
+	public void addTopic(String topic){
+		topics.add(topic);
 	}
 
 }

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/protocols/packets/MQTT_packet.java

@@ -1,6 +1,7 @@
 package de.tu_darmstadt.tk.SmartHomeNetworkSim.core.protocols.packets;
 
 import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Packet;
+import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Port;
 
 /**
  * Packet generated by the MQTT Protocol
@@ -10,14 +11,132 @@ import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Packet;
  */
 public class MQTT_packet extends Packet {
 
+	/*
+	 * List of the different MQTT packet types, which are the bits 7 to 4 of the
+	 * first byte of the packet as specified in MQTT Specification 2.2.1 (MQTT
+	 * Control Packet type)<br> The first 4 bits contain the values 0 to F.
+	 */
+	/*
+	 * The second part of the first byte are the Flag Bits. In most control
+	 * packets they are 0. The Publish Packet contains DUP, QoS, QoS, RETAIN
+	 * bits. PUBREL, SUBSCRIBE and UNSUBSCRIBE contain 0x2 as lower nibble. See
+	 * MQTT Section 2.2.2 for further infos.
+	 */
+
+	/**
+	 * Reserved (forbidden)
+	 */
+	public static final byte RESERVED = 0x00;
+	/**
+	 * Client request to connect to Server (Client->Server)
+	 */
+	public static final byte CONNECT = (byte) 0x10;
+	/**
+	 * Connect acknowledgement (Client<-Server)
+	 */
+	public static final byte CONNACK = (byte) 0x20;
+	/**
+	 * Publish message (Client<->Server)
+	 */
+	public static final byte PUBLISH = (byte) 0x30;
+	/**
+	 * Publish acknowledgement (Client<->Server)
+	 */
+	public static final byte PUBACK = (byte) 0x40;
+	/**
+	 * Publish received (assured delivery part 1) (Client<->Server)
+	 */
+	public static final byte PUBREC = (byte) 0x50;
+	/**
+	 * Publish received (assured delivery part 2) (Client<->Server)
+	 */
+	public static final byte PUBREL = (byte) 0x62;
+	/**
+	 * Publish received (assured delivery part 3) (Client<->Server)
+	 */
+	public static final byte PUBCOMP = (byte) 0x70;
+	/**
+	 * Client subscribe request (Client->Server)
+	 */
+	public static final byte SUBSCRIBE = (byte) 0x82;
+	/**
+	 * Subscribe acknowledgement (Client<-Server)
+	 */
+	public static final byte SUBACK = (byte) 0x90;
+	/**
+	 * Unsubscribe request (Client->Server)
+	 */
+	public static final byte UNSUBSCRIBE = (byte) 0xA2;
+	/**
+	 * Unsubscribe acknowledgement (Client<-Server)
+	 */
+	public static final byte UNSUBACK = (byte) 0xB0;
+	/**
+	 * PING request (Client->Server)
+	 */
+	public static final byte PINGREQ = (byte) 0xC0;
+	/**
+	 * PING response (Client<-Server)
+	 */
+	public static final byte PINGRESP = (byte) 0xD0;
+	/**
+	 * Client is disconnecting (Client->Server)
+	 */
+	public static final byte DISCONNECT = (byte) 0xE0;
+	/**
+	 * Reserved - forbidden
+	 */
+	public static final byte RESERVED2 = (byte) 0xF0;
+
+	/*
+	 * Packet Content
+	 */
+	/**
+	 * First byte of the MQTT:Packet
+	 */
+	protected byte controlByte;
+
+	/**
+	 * Payload message
+	 */
+	protected String message = "";
+
 	/**
 	 * Creates a new MQTT Packet
-	 * @param timestamp time the packet was sent
+	 * 
+	 * @param controlType
+	 *            specifies the packetType. one of CONNECT, CONNACK, ...
+	 * @param timestamp
+	 *            time the packet was sent
+	 * @param source
+	 *            source port of the package
+	 * @param destination
+	 *            destination port of the packet
 	 */
-	public MQTT_packet(long timestamp) {
-		super(timestamp);
+	public MQTT_packet(byte controlType, long timestamp, Port source, Port destination) {
+		super(timestamp, source, destination);
+		controlByte = controlType;
 	}
 	
+	/**
+	 * Creates a new MQTT Packet
+	 * 
+	 * @param controlType
+	 *            specifies the packetType. one of CONNECT, CONNACK, ...
+	 * @param timestamp
+	 *            time the packet was sent
+	 * @param source
+	 *            source port of the package
+	 * @param destination
+	 *            destination port of the packet
+	 * @param msg Message of the packet
+	 */
+	public MQTT_packet(byte controlType, long timestamp, Port source, Port destination, String msg) {
+		super(timestamp, source, destination);
+		controlByte = controlType;
+		message = msg;
+	}
+
 	@Override
 	public byte[] dumpBytes() {
 		return "not implemented".getBytes();
@@ -25,11 +144,62 @@ public class MQTT_packet extends Packet {
 
 	@Override
 	public String getTextualRepresentation() {
-		return "[Packet: MQTT]";
+		/**
+		 * Control type of the MQTT Packet as String
+		 */
+		String packetType;
+		switch (controlByte) {
+		case CONNECT:
+			packetType="CONNECT";
+			break;
+		case CONNACK:
+			packetType="CONNACK";
+			break;
+		case PUBLISH:
+			packetType="PUBLISH";
+			break;
+		case PUBACK:
+			packetType="PUBACK";
+			break;
+		case PUBREC:
+			packetType="PUBREC";
+			break;
+		case PUBREL:
+			packetType="PUBREL";
+			break;
+		case PUBCOMP:
+			packetType="PUBCOMP";
+			break;
+		case SUBSCRIBE:
+			packetType="SUBSCRIBE";
+			break;
+		case SUBACK:
+			packetType="SUBACK";
+			break;
+		case UNSUBSCRIBE:
+			packetType="UNSUBSCRIBE";
+			break;
+		case UNSUBACK:
+			packetType="UNSUBACK";
+			break;
+		case PINGREQ:
+			packetType="PINGREQ";
+			break;
+		case PINGRESP:
+			packetType="PINGRESP";
+			break;
+		case DISCONNECT:
+			packetType="DISCONNECT";
+			break;
+		default:
+			packetType="RESERVED";
+			break;
+		}
+		return "[MQTT: "+packetType+"; time:"+timestamp+"; source:"+source.getOwner().getName()+"; destination:"+destination.getOwner().getName()+(message==""?"":"; "+message)+"]";
 	}
 
 	@Override
 	public String getPayload() {
-		return "{Test: this}";
+		return "null";
 	}
 }

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/simpleImplementation/SimpleConnection.java

@@ -80,16 +80,16 @@ public class SimpleConnection implements Connection {
 		if(status != ACTIVE)return list;
 		//Generate packets by source
 		if(source.getLastTrigger()+source.getTriggerInterval()<startTime && source.getStatus()==Port.SENDING){
-			list.add(p.generateNextPaket(source, startTime,Math.random()<packetLossProbability));
+			list.addAll(p.generateNextPakets(source, startTime,Math.random()<packetLossProbability));
 		}
 		while(source.getLastTrigger()+source.getTriggerInterval()<startTime+duration &&source.getStatus()==Port.SENDING)
-			list.add(p.generateNextPaket(source, source.getLastTrigger()+source.getTriggerInterval(),Math.random()<packetLossProbability));
+			list.addAll(p.generateNextPakets(source, source.getLastTrigger()+source.getTriggerInterval(),Math.random()<packetLossProbability));
 		
 		if(destination.getLastTrigger()+destination.getTriggerInterval()<startTime && destination.getStatus()==Port.SENDING){
-			list.add(p.generateNextPaket(destination, startTime,Math.random()<packetLossProbability));
+			list.addAll(p.generateNextPakets(destination, startTime,Math.random()<packetLossProbability));
 		}
 		while(destination.getLastTrigger()+destination.getTriggerInterval()<startTime+duration &&destination.getStatus()==Port.SENDING)
-			list.add(p.generateNextPaket(destination, destination.getLastTrigger()+destination.getTriggerInterval(),Math.random()<packetLossProbability));
+			list.addAll(p.generateNextPakets(destination, destination.getLastTrigger()+destination.getTriggerInterval(),Math.random()<packetLossProbability));
 
 		list.sort((x,y) -> Long.compare(x.getTimestamp(),y.getTimestamp()));
 		return list;
@@ -120,9 +120,8 @@ public class SimpleConnection implements Connection {
 	public Collection<Packet> getTerminationPackages(long startTime) {
 		status = DONE;
 		return new LinkedList<Packet>(Arrays.asList((Packet) new SimplePacket(
-				startTime, srcOfTermination.getOwner(),
-				srcOfTermination == source ? destination.getOwner() : source
-						.getOwner(), "Terminated")));
+				startTime, srcOfTermination,
+				srcOfTermination == source ? destination : source, "Terminated")));
 	}
 
 	@Override

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/simpleImplementation/SimplePacket.java

@@ -1,6 +1,7 @@
 package de.tu_darmstadt.tk.SmartHomeNetworkSim.core.simpleImplementation;
 
 import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Packet;
+import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Port;
 import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.SmartDevice;
 
 /**
@@ -9,22 +10,6 @@ import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.SmartDevice;
  * @author Andreas T. Meyer-Berg
  */
 public class SimplePacket extends Packet {
-
-	/**
-	 * Time the packet was sent
-	 */
-	private long time;
-
-	/**
-	 * Source of this packet
-	 */
-	private SmartDevice source;
-
-	/**
-	 * Destination of this packet
-	 */
-	private SmartDevice destination;
-
 	/**
 	 * Payload which was sent
 	 */
@@ -40,10 +25,8 @@ public class SimplePacket extends Packet {
 	 * @param destination
 	 *            SmartDevice which should receive this packet
 	 */
-	public SimplePacket(long time, SmartDevice source, SmartDevice destination) {
-		super(time);
-		this.source = source;
-		this.destination = destination;
+	public SimplePacket(long time, Port source, Port destination) {
+		super(time, source, destination);
 		this.payload = "";
 	}
 
@@ -60,8 +43,8 @@ public class SimplePacket extends Packet {
 	 *            String which represents the payload which is encapsulated by
 	 *            this packet
 	 */
-	public SimplePacket(long time, SmartDevice source, SmartDevice destination, String payload) {
-		super(time);
+	public SimplePacket(long time, Port source, Port destination, String payload) {
+		super(time, source, destination);
 		this.source = source;
 		this.destination = destination;
 		this.payload = payload;
@@ -79,10 +62,10 @@ public class SimplePacket extends Packet {
 	
 	@Override
 	public String toString() {
-		String destName = destination == null ? "null" : destination.getName();
-		String srcName = source == null ? "null" : source.getName();
+		String destName = destination == null ? "null" : destination.getOwner().getName();
+		String srcName = source == null ? "null" : source.getOwner().getName();
 			
-		return "[SimplePacket:" + payload + " time-" + time + ";source:" + srcName + ";dest:"
+		return "[SimplePacket:" + payload + " time-" + timestamp + ";source:" + srcName + ";dest:"
 				+ destName + "]";
 	}
 

src/main/java/de/tu_darmstadt/tk/SmartHomeNetworkSim/core/simpleImplementation/SimpleProtocol.java

@@ -2,6 +2,7 @@ package de.tu_darmstadt.tk.SmartHomeNetworkSim.core.simpleImplementation;
 
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.LinkedList;
 
 import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Packet;
 import de.tu_darmstadt.tk.SmartHomeNetworkSim.core.Port;
@@ -30,13 +31,15 @@ public class SimpleProtocol implements Protocol {
 		srcSends = true;
 	}
 	@Override
-	public Packet generateNextPaket(Port p, long timestep, boolean packetLost) {
+	public Collection<Packet> generateNextPakets(Port p, long timestep, boolean packetLost) {
+		LinkedList<Packet> ret = new LinkedList<Packet>();
 		p.setLastTrigger(timestep);
-		if(packetLost)return new SimplePacket(timestep, p.getOwner(), p==source?destination.getOwner():source.getOwner(), "Lost");
+		if(packetLost)ret.add(new SimplePacket(timestep, p, p==source?destination:source, "Lost"));
 		if(p == destination)
-			return new SimplePacket(timestep, destination.getOwner(), source.getOwner());
+			ret.add(new SimplePacket(timestep, destination, source));
 		else
-			return new SimplePacket(timestep, source.getOwner(), destination.getOwner());
+			ret.add(new SimplePacket(timestep, source, destination));
+		return ret;
 	}
 	
 	@Override