Initial commit

.classpath

+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/classes" path="src/main/resources">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="test" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry excluding="**" kind="src" output="target/test-classes" path="src/test/resources">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+			<attribute name="test" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-11">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>

.project

+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>pb</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+	</natures>
+</projectDescription>

.settings/org.eclipse.core.resources.prefs

+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/main/resources=UTF-8
+encoding//src/test/java=UTF-8
+encoding//src/test/resources=UTF-8
+encoding/<project>=UTF-8

.settings/org.eclipse.jdt.core.prefs

+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=11
+org.eclipse.jdt.core.compiler.compliance=11
+org.eclipse.jdt.core.compiler.problem.enablePreviewFeatures=disabled
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.problem.reportPreviewFeatures=ignore
+org.eclipse.jdt.core.compiler.release=disabled
+org.eclipse.jdt.core.compiler.source=11

.settings/org.eclipse.m2e.core.prefs

+activeProfiles=
+eclipse.preferences.version=1
+resolveWorkspaceProjects=true
+version=1

pom.xml

+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <groupId>unimelb</groupId>
+  <artifactId>pb</artifactId>
+  <version>0.0.1-SNAPSHOT</version>
+  
+  <properties>
+    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <maven.compiler.source>11</maven.compiler.source>
+    <maven.compiler.target>11</maven.compiler.target>
+  </properties>
+
+  <dependencies>
+	<dependency>
+	    <groupId>com.googlecode.json-simple</groupId>
+	    <artifactId>json-simple</artifactId>
+	    <version>1.1.1</version>
+	</dependency>
+	<dependency>
+    	<groupId>commons-cli</groupId>
+    	<artifactId>commons-cli</artifactId>
+    	<version>1.3.1</version>
+    </dependency>
+  </dependencies>
+  
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-assembly-plugin</artifactId>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>single</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <descriptorRefs>
+            <descriptorRef>jar-with-dependencies</descriptorRef>
+          </descriptorRefs>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>
+  
+</project>
\ No newline at end of file

src/main/java/pb/Client.java

+package pb;
+
+import java.io.IOException;
+import java.util.logging.Logger;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+
+import pb.client.ClientManager;
+
+/**
+ * Client main. Parse command line options and provide default values.
+ * 
+ * @see {@link pb.ClientManager}
+ * @see {@link pb.Utils}
+ * @author aaron
+ *
+ */
+public class Client  {
+	private static Logger log = Logger.getLogger(Client.class.getName());
+	private static int port=Utils.serverPort; // default port number for the server
+	private static String host=Utils.serverHost; // default host for the server
+	
+	private static void help(Options options){
+		String header = "PB Client for Unimelb COMP90015\n\n";
+		String footer = "\ncontact aharwood@unimelb.edu.au for issues.";
+		HelpFormatter formatter = new HelpFormatter();
+		formatter.printHelp("pb.Client", header, options, footer, true);
+		System.exit(-1);
+	}
+	
+	public static void main( String[] args ) throws IOException
+    {
+    	// set a nice log format
+		System.setProperty("java.util.logging.SimpleFormatter.format",
+                "[%1$tl:%1$tM:%1$tS:%1$tL] %2$s %4$s: %5$s%n");
+        
+    	// parse command line options
+        Options options = new Options();
+        options.addOption("port",true,"server port, an integer");
+        options.addOption("host",true,"hostname, a string");
+        
+        CommandLineParser parser = new DefaultParser();
+        CommandLine cmd = null;
+        try {
+			cmd = parser.parse( options, args);
+		} catch (ParseException e1) {
+			help(options);
+		}
+        
+        if(cmd.hasOption("port")){
+        	try{
+        		port = Integer.parseInt(cmd.getOptionValue("port"));
+			} catch (NumberFormatException e){
+				System.out.println("-port requires a port number, parsed: "+cmd.getOptionValue("port"));
+				help(options);
+			}
+        }
+        
+        if(cmd.hasOption("host")) {
+        	host = cmd.getOptionValue("host");
+        }
+        
+        // start up the client
+        log.info("PB Client starting up");
+        
+        // the client manager will make a connection with the server
+        // and the connection will use a thread that prevents the JVM
+        // from terminating immediately
+        new ClientManager(host,port);
+        
+    }
+}

src/main/java/pb/Endpoint.java

+package pb;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.logging.Logger;
+
+import pb.protocols.InvalidMessage;
+import pb.protocols.Message;
+import pb.protocols.Protocol;
+import pb.protocols.IRequestReplyProtocol;
+import pb.protocols.keepalive.KeepAliveProtocol;
+import pb.protocols.session.SessionProtocol;
+
+/**
+ * The endpoint is a thread that blocking reads incoming messages (on a socket)
+ * and sends them to the appropriate protocol for processing; thus a
+ * thread-per-connection model is being used. It also provides a synchronized
+ * method to send data to the socket which will be sent to the other endpoint.
+ * Any number of protocols can be handled by the endpoint, but there can be only
+ * one instance of each protocol running at a time.
+ * 
+ * @see {@link pb.Manager}
+ * @see {@link pb.protocols.session.SessionProtocol}
+ * @see {@link pb.protocols.keepalive.KeepAliveProtocol}
+ * @author aaron
+ *
+ */
+public class Endpoint extends Thread {
+	private static Logger log = Logger.getLogger(Endpoint.class.getName());
+	
+	/**
+	 * The socket this endpoint is wrapped around.
+	 */
+	private Socket socket;
+	
+	/**
+	 * The manager to report to when things happen.
+	 */
+	private Manager manager;
+	
+	/**
+	 * The input data stream on the socket.
+	 */
+	private DataInputStream in=null;
+	
+	/**
+	 * The output data stream on the socket.
+	 */
+	private DataOutputStream out=null;
+	
+	/**
+	 * A protocol name to protocol map, of protocols in use.
+	 */
+	private Map<String,Protocol> protocols;
+	
+	/**
+	 * Initialise the endpoint with a socket and a manager.
+	 * @param socket
+	 * @param manager
+	 */
+	public Endpoint(Socket socket, Manager manager) {
+		this.socket = socket;
+		this.manager = manager;
+		protocols = new HashMap<>();
+	}
+	
+	/**
+	 * Send a Message on the socket for this endpoint. This is synchronized
+	 * to avoid multiple concurrent messages overwriting each other on the socket.
+	 * @param msg
+	 * @return true if the message was sent, false otherwise
+	 * @throws EndpointUnavailable if the endpoint is not yet ready 
+	 * or if the endpoint is terminated
+	 */
+	public synchronized boolean send(Message msg) throws EndpointUnavailable {
+		if(out==null) {
+			throw new EndpointUnavailable();
+		}
+		try {
+			log.info("sending "+msg.getName()+" for protocol "+msg.getProtocolName()+" to "+getOtherEndpointId());
+			out.writeUTF(msg.toJsonString());
+			out.flush();
+		} catch (IOException e) {
+			manager.endpointDisconnectedAbruptly(this);
+			return false;
+		}
+		return true;
+	}
+	
+	/**
+	 * Closes the endpoint, which closes the socket
+	 */
+	public synchronized void close() {
+		// make sure all of the protocols have stopped
+		Set<String> protocolNames;
+		synchronized(protocols) {
+			protocolNames = new HashSet<String>(protocols.keySet());
+		}
+		if(protocolNames!=null)
+			protocolNames.forEach((protocolName)->{stopProtocol(protocolName);});
+		interrupt();
+		try {
+			if(out!=null) out.close();
+			out=null;
+		} catch (IOException e) {
+			log.warning("connection did not close properly: "+e.getMessage());
+		}
+		try {
+			socket.close();
+		} catch (IOException e) {
+			log.warning("socket did not close properly: "+e.getMessage());
+		}
+		manager.endpointClosed(this);
+	}
+	
+	/**
+	 * Continue to read messages from the socket until interrupted.
+	 */
+	@Override
+	public void run() {
+		try {
+			in = new DataInputStream(socket.getInputStream());
+			out = new DataOutputStream(socket.getOutputStream());
+		} catch (IOException e){
+			manager.endpointDisconnectedAbruptly(this);
+			return;
+		}
+		manager.endpointReady(this);
+		log.info("endpoint has started to: "+getOtherEndpointId());
+		while(!isInterrupted()) {
+			try {
+				String line=in.readUTF();
+				Message msg = Message.toMessage(line);
+				Protocol protocol=null;
+				synchronized(protocols) {
+					protocol=protocols.get(msg.getProtocolName());
+				}
+				if(protocol==null) {
+					switch(msg.getProtocolName()) {
+					case SessionProtocol.protocolName:
+						protocol=new SessionProtocol(this,manager);
+						break;
+					case KeepAliveProtocol.protocolName:
+						protocol=new KeepAliveProtocol(this,manager);
+					}
+					if(!manager.protocolRequested(this,protocol)) {
+						log.info("message dropped due to no protocol available: "+line);
+						continue;
+					}
+				}
+				log.info("received "+msg.getName()+" for protocol "+msg.getProtocolName()+" from "+getOtherEndpointId());
+				switch(msg.getType()) {
+				case Request:
+					((IRequestReplyProtocol)protocol).receiveRequest(msg);
+					break;
+				case Reply:
+					((IRequestReplyProtocol)protocol).receiveReply(msg);
+					break;
+				}
+			} catch (IOException e) {
+				manager.endpointDisconnectedAbruptly(this);
+				// we can't continue here
+				break;
+			} catch (InvalidMessage e) {
+				manager.endpointSentInvalidMessage(this);
+				// up to the client what to do
+			}catch (EndpointUnavailable e) {
+				manager.endpointDisconnectedAbruptly(this);
+				break;
+			}
+		}
+		try {
+			in.close();
+		} catch (IOException e) {
+			log.warning("connection did not close properly: "+e.getMessage());
+		}
+		log.info("endpoint has terminated to: "+getOtherEndpointId());
+	}
+	
+	/**
+	 * Start handling a protocol. Only one instance of a protocol can be handled
+	 * at a time. Either client or server may start/initiate the use of the protocol.
+	 * @see {@link pb.protocols.Protocol}
+	 * @param protocol the protocol to handle
+	 * @throws ProtocolAlreadyRunning if there is already an instance of this protocol
+	 * running on this endpoint
+	 */
+	public void handleProtocol(Protocol protocol) throws ProtocolAlreadyRunning {
+		synchronized(protocols) {
+			if(protocols.containsKey(protocol.getProtocolName())){
+				throw new ProtocolAlreadyRunning();
+			} else {
+				protocols.put(protocol.getProtocolName(),protocol);
+				log.info("now handling protocol: "+protocol.getProtocolName());
+			}
+		}
+	}
+	
+	/**
+	 * Stop a protocol that is already being handled. It will be removed
+	 * from the endpoints set of handled protocols.
+	 * @see {@link pb.protocols.Protocol}
+	 * @param protocolName the protocol name to stop
+	 */
+	public void stopProtocol(String protocolName) {
+		synchronized(protocols) {
+			if(!protocols.containsKey(protocolName)) {
+				log.warning("no instance of protocol to stop: "+protocolName);
+				return;
+			}
+			protocols.get(protocolName).stopProtocol();
+			protocols.remove(protocolName);
+		}
+	}
+	
+	/**
+	 * 
+	 * @return the id of the other endpoint
+	 */
+	public String getOtherEndpointId() {
+		return socket.getInetAddress().toString()+":"+socket.getPort();
+	}
+}

src/main/java/pb/EndpointUnavailable.java

+package pb;
+
+@SuppressWarnings("serial")
+public class EndpointUnavailable extends Exception {
+
+}

src/main/java/pb/Manager.java

+package pb;
+
+
+import pb.protocols.Protocol;
+
+/**
+ * Manager base class. Methods must be overriden.
+ * 
+ * @see {@link pb.server.ServerManager}
+ * @see {@link pb.client.ClientManager}
+ * @author aaron
+ *
+ */
+public class Manager {
+	
+	/**
+	 * The endpoint is ready to use.
+	 * @param endpoint
+	 */
+	public void endpointReady(Endpoint endpoint) {
+		
+	}
+	
+	/**
+	 * The endpoint close() method has been called and completed.
+	 * @param endpoint
+	 */
+	public void endpointClosed(Endpoint endpoint) {
+		
+	}
+	
+	/**
+	 * The endpoint has abruptly disconnected. It can no longer
+	 * send or receive data.
+	 * @param endpoint
+	 */
+	public void endpointDisconnectedAbruptly(Endpoint endpoint) {
+		
+	}
+
+	/**
+	 * An invalid message was received over the endpoint.
+	 * @param endpoint
+	 */
+	public void endpointSentInvalidMessage(Endpoint endpoint) {
+		
+	}
+	
+
+	/**
+	 * The protocol on the endpoint is not responding.
+	 * @param endpoint
+	 */
+	public void endpointTimedOut(Endpoint endpoint,Protocol protocol) {
+		
+	}
+
+	/**
+	 * The protocol on the endpoint has been violated.
+	 * @param endpoint
+	 */
+	public void protocolViolation(Endpoint endpoint,Protocol protocol) {
+		
+	}
+
+	/**
+	 * The session protocol is indicating that a session has started.
+	 * @param endpoint
+	 */
+	public void sessionStarted(Endpoint endpoint) {
+		
+	}
+
+	/**
+	 * The session protocol is indicating that the session has stopped. 
+	 * @param endpoint
+	 */
+	public void sessionStopped(Endpoint endpoint) {
+		
+	}
+	
+
+	/**
+	 * The endpoint has requested a protocol to start. If the protocol
+	 * is allowed then the manager should tell the endpoint to handle it
+	 * using {@link pb.Endpoint#handleProtocol(Protocol)}
+	 * before returning true.
+	 * @param protocol
+	 * @return true if the protocol was started, false if not (not allowed to run)
+	 */
+	public boolean protocolRequested(Endpoint endpoint, Protocol protocol) {
+		return false;
+	}
+
+	
+
+
+}

src/main/java/pb/ProtocolAlreadyRunning.java

+package pb;
+
+@SuppressWarnings("serial")
+public class ProtocolAlreadyRunning extends Exception {
+
+}

src/main/java/pb/Server.java

+package pb;
+
+import java.io.IOException;
+import java.util.logging.Logger;
+
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.DefaultParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+
+import pb.server.ServerManager;
+
+/**
+ * Server main. Parse command line options and provide default values.
+ * 
+ * @see {@link pb.ServerManager}
+ * @see {@link pb.Utils}
+ * @author aaron
+ *
+ */
+public class Server {
+	private static Logger log = Logger.getLogger(Server.class.getName());
+	private static int port=Utils.serverPort; // default port number for the server
+	
+
+	private static void help(Options options){
+		String header = "PB Server for Unimelb COMP90015\n\n";
+		String footer = "\ncontact aharwood@unimelb.edu.au for issues.";
+		HelpFormatter formatter = new HelpFormatter();
+		formatter.printHelp("pb.Server", header, options, footer, true);
+		System.exit(-1);
+	}
+	
+	public static void main( String[] args ) throws IOException
+    {
+    	// set a nice log format
+		System.setProperty("java.util.logging.SimpleFormatter.format",
+                "[%1$tl:%1$tM:%1$tS:%1$tL] %2$s %4$s: %5$s%n");
+        
+    	// parse command line options
+        Options options = new Options();
+        options.addOption("port",true,"server port, an integer");
+        
+       
+        CommandLineParser parser = new DefaultParser();
+        CommandLine cmd = null;
+        try {
+			cmd = parser.parse( options, args);
+		} catch (ParseException e1) {
+			help(options);
+		}
+        
+        if(cmd.hasOption("port")){
+        	try{
+        		port = Integer.parseInt(cmd.getOptionValue("port"));
+			} catch (NumberFormatException e){
+				System.out.println("-port requires a port number, parsed: "+cmd.getOptionValue("port"));
+				help(options);
+			}
+        }
+        
+        
+        // start up the server
+        log.info("PB Server starting up");
+        
+        // the server manager will start an io thread and this will prevent
+        // the JVM from terminating
+        new ServerManager(port);
+        
+    }
+}

src/main/java/pb/Utils.java

+package pb;
+
+import java.util.Timer;
+import java.util.TimerTask;
+
+import pb.protocols.ICallback;
+
+/**
+ * A singleton class to provide various utility functions. It must always be
+ * accessed statically as Utils.getInstance()...
+ * 
+ * @author aaron
+ *
+ */
+public class Utils {
+	private static Utils utils;
+	
+	/**
+	 * Default server port
+	 */
+	public static final int serverPort = 3100;
+	
+	/**
+	 * Default server host
+	 */
+	public static final String serverHost = "localhost";
+	
+	
+	/**
+	 * Use of a single timer object over the entire system helps
+	 * to reduce thread usage.
+	 */
+	Timer timer = new Timer();
+	
+	public Utils() {
+		timer=new Timer();
+	}
+	
+	public static synchronized Utils getInstance() {
+		if(utils==null) utils=new Utils();
+		return utils;
+	}
+	
+	/**
+	 * Convenience method to set an anonymous method callback
+	 * after a timeout delay. Go JavaScript :-)
+	 * <br/>
+	 * Use this method like: 
+	 * <code>
+	 * Utils.getInstance().setTimeout(()->{doSomething();},10000);
+	 * </code>
+	 * @param callback the method to call
+	 * @param delay the delay in ms before calling the method
+	 */
+	public void setTimeout(ICallback callback,long delay) {
+		
+		timer.schedule(new TimerTask() {
+			@Override
+			public void run() {
+				callback.callback();
+			}
+			
+		}, delay);
+	}
+	
+	/**
+	 * Call before the system exits.
+	 */
+	public void cleanUp() {
+		timer.cancel();
+	}
+}

src/main/java/pb/client/ClientManager.java

+package pb.client;
+
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.util.logging.Logger;
+
+import pb.Endpoint;
+import pb.EndpointUnavailable;
+import pb.Manager;
+import pb.ProtocolAlreadyRunning;
+import pb.Utils;
+import pb.protocols.IRequestReplyProtocol;
+import pb.protocols.Protocol;
+import pb.protocols.keepalive.KeepAliveProtocol;
+import pb.protocols.session.SessionProtocol;
+
+/**
+ * Manages the connection to the server and the client's state.
+ * 
+ * @see {@link pb.Manager}
+ * @see {@link pb.Endpoint}
+ * @see {@link pb.protocols.Protocol}
+ * @see {@link pb.protocols.IRequestReplyProtocol}
+ * @author aaron
+ *
+ */
+public class ClientManager extends Manager {
+	private static Logger log = Logger.getLogger(ClientManager.class.getName());
+	private SessionProtocol sessionProtocol;
+	private KeepAliveProtocol keepAliveProtocol;
+	private Socket socket;
+	
+	public ClientManager(String host,int port) throws UnknownHostException, IOException {
+		
+		socket=new Socket(InetAddress.getByName(host),port);
+		Endpoint endpoint = new Endpoint(socket,this);
+		endpoint.start();
+		
+		// simulate the client shutting down after 2mins
+		// this will be removed when the client actually does something
+		// controlled by the user
+		Utils.getInstance().setTimeout(()->{
+			try {
+				sessionProtocol.stopSession();
+			} catch (EndpointUnavailable e) {
+				//ignore...
+			}
+		}, 120000);
+		
+		
+		try {
+			// just wait for this thread to terminate
+			endpoint.join();
+		} catch (InterruptedException e) {
+			// just make sure the ioThread is going to terminate
+			endpoint.close();
+		}
+		
+		Utils.getInstance().cleanUp();
+	}
+	
+	/**
+	 * The endpoint is ready to use.
+	 * @param endpoint
+	 */
+	@Override
+	public void endpointReady(Endpoint endpoint) {
+		log.info("connection with server established");
+		sessionProtocol = new SessionProtocol(endpoint,this);
+		try {
+			// we need to add it to the endpoint before starting it
+			endpoint.handleProtocol(sessionProtocol);
+			sessionProtocol.startAsClient();
+		} catch (EndpointUnavailable e) {
+			log.severe("connection with server terminated abruptly");
+			endpoint.close();
+		} catch (ProtocolAlreadyRunning e) {
+			// hmmm, so the server is requesting a session start?
+			log.warning("server initiated the session protocol... weird");
+		}
+		keepAliveProtocol = new KeepAliveProtocol(endpoint,this);
+		try {
+			// we need to add it to the endpoint before starting it
+			endpoint.handleProtocol(keepAliveProtocol);
+			keepAliveProtocol.startAsClient();
+		} catch (EndpointUnavailable e) {
+			log.severe("connection with server terminated abruptly");
+			endpoint.close();
+		} catch (ProtocolAlreadyRunning e) {
+			// hmmm, so the server is requesting a session start?
+			log.warning("server initiated the session protocol... weird");
+		}
+	}
+	
+	/**
+	 * The endpoint close() method has been called and completed.
+	 * @param endpoint
+	 */
+	public void endpointClosed(Endpoint endpoint) {
+		log.info("connection with server terminated");
+	}
+	
+	/**
+	 * The endpoint has abruptly disconnected. It can no longer
+	 * send or receive data.
+	 * @param endpoint
+	 */
+	@Override
+	public void endpointDisconnectedAbruptly(Endpoint endpoint) {
+		log.severe("connection with server terminated abruptly");
+		endpoint.close();
+	}
+
+	/**
+	 * An invalid message was received over the endpoint.
+	 * @param endpoint
+	 */
+	@Override
+	public void endpointSentInvalidMessage(Endpoint endpoint) {
+		log.severe("server sent an invalid message");
+		endpoint.close();
+	}
+	
+
+	/**
+	 * The protocol on the endpoint is not responding.
+	 * @param endpoint
+	 */
+	@Override
+	public void endpointTimedOut(Endpoint endpoint,Protocol protocol) {
+		log.severe("server has timed out");
+		endpoint.close();
+	}
+
+	/**
+	 * The protocol on the endpoint has been violated.
+	 * @param endpoint
+	 */
+	@Override
+	public void protocolViolation(Endpoint endpoint,Protocol protocol) {
+		log.severe("protocol with server has been violated: "+protocol.getProtocolName());
+		endpoint.close();
+	}
+
+	/**
+	 * The session protocol is indicating that a session has started.
+	 * @param endpoint
+	 */
+	@Override
+	public void sessionStarted(Endpoint endpoint) {
+		log.info("session has started with server");
+		
+		// we can now start other protocols with the server
+	}
+
+	/**
+	 * The session protocol is indicating that the session has stopped. 
+	 * @param endpoint
+	 */
+	@Override
+	public void sessionStopped(Endpoint endpoint) {
+		log.info("session has stopped with server");
+		endpoint.close(); // this will stop all the protocols as well
+	}
+	
+
+	/**
+	 * The endpoint has requested a protocol to start. If the protocol
+	 * is allowed then the manager should tell the endpoint to handle it
+	 * using {@link pb.Endpoint#handleProtocol(Protocol)}
+	 * before returning true.
+	 * @param protocol
+	 * @return true if the protocol was started, false if not (not allowed to run)
+	 */
+	@Override
+	public boolean protocolRequested(Endpoint endpoint, Protocol protocol) {
+		// the only protocols in this system are this kind...
+		try {
+			((IRequestReplyProtocol)protocol).startAsClient();
+			endpoint.handleProtocol(protocol);
+			return true;
+		} catch (EndpointUnavailable e) {
+			// very weird... should log this
+			return false;
+		} catch (ProtocolAlreadyRunning e) {
+			// even more weird... should log this too
+			return false;
+		}
+	}
+
+}

src/main/java/pb/protocols/Document.java

+package pb.protocols;
+
+import java.util.ArrayList;
+
+import org.json.simple.JSONArray;
+import org.json.simple.JSONObject;
+import org.json.simple.parser.JSONParser;
+import org.json.simple.parser.ParseException;
+
+/**
+ * Helper class for using JSON. Example usage:
+ * <pre>
+ * {@code
+ * Document doc1 = new Document();
+ * doc1.append("host","localhost");
+ * doc1.append("port",8111);
+ * String host = doc1.getString("host");
+ * int port = doc1.getInteger("port");
+ * String json1 = doc1.toJson(); // convert Document to a JSON String
+ * Document doc2 = Document.parse(json1); // convert JSON String back to Document
+ * ArrayList<Document> docs = new ArrayList<Document>();
+ * docs.add(doc1);
+ * docs.add(doc2);
+ * Document doc3 = new Document();
+ * doc3.append("docList",docs);
+ * doc3.toJson(); // {"docList":[{"host":"localhost","port":8111},{"host":"localhost","port":8111}]}
+ * ArrayList<Document> docs2 = (ArrayList<Document>) doc3.get("docList");
+ * }
+ * </pre>
+ * @author aaron
+ *
+ */
+public class Document {
+	
+	protected JSONObject obj;
+	
+	public Document(){
+		obj=new JSONObject();
+	}
+	
+	public Document(JSONObject obj){
+		this.obj = obj;
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,String val){
+		if(val==null){
+			obj.put(key, null);
+		} else {
+			obj.put(key, new String(val));
+		}
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,Document doc){
+		obj.put(key, doc.obj);
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,boolean val){
+		obj.put(key, Boolean.valueOf(val));
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,ArrayList<?> val){
+		JSONArray list = new JSONArray();
+		for(Object o : val){
+			if(o instanceof Document){
+				list.add(((Document)o).obj);
+			} else {
+				list.add(o);
+			}
+		}
+		obj.put(key,list);
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,long val){
+		obj.put(key, Long.valueOf(val));
+	}
+	
+	@SuppressWarnings("unchecked")
+	public void append(String key,int val){
+		obj.put(key, Integer.valueOf(val));
+	}
+	
+	public String toJson(){
+		return obj.toJSONString();
+	}
+	
+	public static Document parse(String json) {
+		JSONParser parser = new JSONParser();
+		try {
+			JSONObject obj  = (JSONObject) parser.parse(json);
+			return new Document(obj);
+		} catch (ParseException e) {
+			return new Document();
+		} catch (ClassCastException e){
+			return new Document();
+		}
+	}
+	
+	public boolean containsKey(String key){
+		return obj.containsKey(key);
+	}
+	
+	public String getString(String key){
+		return (String) obj.get(key);
+	}
+	
+	private ArrayList<Object> getList(JSONArray o){
+		ArrayList<Object> list = new ArrayList<Object>();
+		for(Object l : (JSONArray)o){
+			if(l instanceof JSONObject){
+				list.add(new Document((JSONObject) l));
+			} else if(l instanceof JSONArray){
+				list.add(getList((JSONArray) l));
+			} else {
+				list.add(l);
+			}
+		}
+		return list;
+	}
+	
+	public Object get(String key){
+		Object o = obj.get(key);
+		if(o instanceof JSONObject){
+			return (Object) new Document((JSONObject) o);
+		} else if(o instanceof JSONArray){
+			return getList((JSONArray)o);
+		} else {
+			return o;
+		}
+		
+	}
+	
+	public int getInteger(String key){
+		return (int) obj.get(key);
+	}
+	
+	public long getLong(String key){
+		return (long) obj.get(key);
+	}
+	
+	public boolean getBoolean(String key){
+		return (boolean) obj.get(key);
+	}
+}

src/main/java/pb/protocols/ICallback.java

+package pb.protocols;
+
+public interface ICallback {
+	public void callback();
+}

src/main/java/pb/protocols/IRequestReplyProtocol.java

+package pb.protocols;
+
+import pb.EndpointUnavailable;
+
+/**
+ * Request/reply protocol objects must implement this interface.
+ * 
+ * @see {@link #startAsClient()}
+ * @see {@link #startAsServer()}
+ * @see {@link #sendRequest(Message)}
+ * @see {@link #receiveReply(Message)}
+ * @see {@link #receiveRequest(Message)}
+ * @see {@link #sendReply(Message)}
+ * @see {@link pb.protocols.Message}
+ * @author aaron
+ *
+ */
+public interface IRequestReplyProtocol {
+	/**
+	 * Start the protocol as a client.
+	 * @throws EndpointUnavailable
+	 */
+	public void startAsClient() throws EndpointUnavailable;
+	
+	/**
+	 * Start the protocol as a server.
+	 * @throws EndpointUnavailable
+	 */
+	public void startAsServer() throws EndpointUnavailable;
+	
+	/**
+	 * Send a request message.
+	 * @param msg
+	 * @throws EndpointUnavailable
+	 */
+	public void sendRequest(Message msg) throws EndpointUnavailable;
+	
+	/**
+	 * Receive a reply message.
+	 * @param msg
+	 * @throws EndpointUnavailable
+	 */
+	public void receiveReply(Message msg) throws EndpointUnavailable;
+	
+	/**
+	 * Receive a request message.
+	 * @param msg
+	 * @throws EndpointUnavailable
+	 */
+	public void receiveRequest(Message msg) throws EndpointUnavailable;
+	
+	/**
+	 * Send a reply message.
+	 * @param msg
+	 * @throws EndpointUnavailable
+	 */
+	public void sendReply(Message msg) throws EndpointUnavailable;
+}

src/main/java/pb/protocols/InvalidMessage.java

+package pb.protocols;
+
+
+/**
+ * The message is not valid. It may be missing required parameters or the
+ * parameters may be of the wrong type. Note that the presence of additional
+ * parameters does not trigger this exception.
+ * 
+ * @author aaron
+ *
+ */
+@SuppressWarnings("serial")
+public class InvalidMessage extends Exception {
+
+}

src/main/java/pb/protocols/Message.java

+package pb.protocols;
+
+import pb.protocols.keepalive.KeepAliveReply;
+import pb.protocols.keepalive.KeepAliveRequest;
+import pb.protocols.session.SessionStartReply;
+import pb.protocols.session.SessionStartRequest;
+import pb.protocols.session.SessionStopReply;
+import pb.protocols.session.SessionStopRequest;
+
+/**
+ * Message super class and factory for all protocol messages, to parse a
+ * received UTF-8 line of text in JSON format, as an object that represents the
+ * message.
+ * 
+ * @see {@link pb.protocols.Protocol}
+ * @author aaron
+ *
+ */
+public class Message {
+	/**
+	 * Messages are either a request or a reply.
+	 */
+	static public enum Type {
+		Request,
+		Reply
+	}
+	
+	/**
+	 * All of the message parameters are wrapped up in a Document class.
+	 */
+	protected Document doc;
+	
+	/**
+	 * Initialiser when given parameters explicitly.
+	 * @param name the name of the message (its classname by convention)
+	 * @param protocolName the name of the protocol the message belongs to
+	 * @param type whether its a Request or a Reply message
+	 */
+	public Message(String name, String protocolName, Message.Type type) {
+		doc = new Document();
+		doc.append("name", name);
+		doc.append("protocolName", protocolName);
+		doc.append("type", type.toString());
+	}
+	
+	static public void validateStringValue(String key,String val,Document doc) throws InvalidMessage {
+		if(!doc.containsKey(key)) throw new InvalidMessage();
+		if(!(doc.get(key) instanceof String)) throw new InvalidMessage();
+		String msg = doc.getString(key);
+		if(!msg.equals(val)) throw new InvalidMessage();
+	}
+	
+	/**
+	 * Initialiser when given parameters in a doc.
+	 * @param name the name of the message that is being initialised
+	 * @param doc with the message details
+	 * @throws InvalidMessage when the name of the message in the doc is incorrect
+	 */
+	public Message(String name, String protocolName, 
+			Message.Type type, Document doc) throws InvalidMessage {
+		validateStringValue("name",name,doc);
+		validateStringValue("protocolName",protocolName,doc);
+		validateStringValue("type",type.toString(),doc);
+	}
+	
+	/**
+	 * Turn a json string into an appropriate message object.
+	 * @param json the string to parse, must be in JSON format
+	 * @return the appropriate message object
+	 * @throws InvalidMessage if no message object matches the message
+	 */
+	static public Message toMessage(String json) throws InvalidMessage {
+		Document doc = Document.parse(json);
+		// the following test is somewhat repetitive, but it avoids having
+		// to test each message type, handling exceptions for those that are
+		// not the matching message type
+		if(!doc.containsKey("name")) throw new InvalidMessage();
+		if(!(doc.get("name") instanceof String)) throw new InvalidMessage();
+		String msg = doc.getString("name");
+		switch(msg) {
+			case KeepAliveRequest.name: return new KeepAliveRequest(doc);
+			case KeepAliveReply.name: return new KeepAliveReply(doc);
+			case SessionStartRequest.name: return new SessionStartRequest(doc);
+			case SessionStartReply.name: return new SessionStartReply(doc);
+			case SessionStopRequest.name: return new SessionStopRequest(doc);
+			case SessionStopReply.name: return new SessionStopReply(doc);
+			// put more message cases here
+			
+			// if nothing matches, its invalid
+			default: throw new InvalidMessage();
+		}
+	}
+	
+	/**
+	 * Convert the message to a string for transmission.
+	 * @return
+	 */
+	public String toJsonString() {
+		return doc.toJson();
+	}
+	
+	/**
+	 * Return the protocol name
+	 * @return
+	 */
+	public String getProtocolName() {
+		return doc.getString("protocolName");
+	}
+	
+	/**
+	 * Return the message name
+	 * @return
+	 */
+	public String getName() {
+		return doc.getString("name");
+	}
+	
+	/**
+	 * Return the message type
+	 * @return
+	 */
+	public Message.Type getType() {
+		return Message.Type.valueOf(doc.getString("type"));
+	}
+	
+}

src/main/java/pb/protocols/Protocol.java

+package pb.protocols;
+
+import pb.Endpoint;
+import pb.Manager;
+
+/**
+ * All protocols have an endpoint and a manager.
+ * @see {@link pb.Endpoint}
+ * @see {@link pb.Manager}
+ * @see {@link pb.protocols.session.SessionProtocol}
+ * @see {@link pb.protocols.keepalive.KeepAliveProtocol}
+ * @author aaron
+ *
+ */
+public class Protocol {
+	/**
+	 * The protocol name is used when routing messages to this protocol.
+	 * It must be unique over all protocols defined.
+	 */
+	public static final String protocolName = "Protocol";
+	
+	/**
+	 * The endpoint that is handling the protocol.
+	 */
+	protected Endpoint endpoint;
+	
+	/**
+	 * The manager to report events to.
+	 */
+	protected Manager manager;
+	
+	/**
+	 * Initialise the protocol with an endpoint and manager.
+	 * @param endpoint
+	 * @param manager
+	 */
+	public Protocol(Endpoint endpoint, Manager manager) {
+		this.endpoint=endpoint;
+		this.manager=manager;
+	}
+	
+	/**
+	 * Signal the protocol to stop. More specifically this method
+	 * is called when the protocol should not undertake any more
+	 * actions, such as processing messages or sending messages.
+	 */
+	public void stopProtocol() {
+		
+	}
+
+	/**
+	 * Sometimes the static string reference is not reachable, so
+	 * this method provides access.
+	 * @return the name of the protocol
+	 */
+	public String getProtocolName() {
+		return protocolName;
+	}
+}