Finish Integration With Relay

Added a system called the Timeline which allows work to be
set to run at specific points in time. This allows work to
be queued.

To allow the relay server data to merge in with the server,
a work item is set to run every minute. When it finishs, it
sets itself to run again in a minute.

When data needs to be sent to the relay, a work item is added
to the timeline so that the current connection can be closed
before sending data to the relay.

The timeline allowed for the Hub to be removed as work items
to handle connections could be added to the timeline.

Closes #4.

Author: vaage

src/codeu/chat/ServerMain.java

@@ -83,30 +83,21 @@ private static void runServer(Uuid id,
 
     final Server server = new Server(id, secret, relay);
 
-    LOG.info("Server object created.");
+    LOG.info("Created server.");
 
-    final Runnable hub = new Hub(serverSource, new Hub.Handler() {
+    while (true) {
 
-      @Override
-      public void handle(Connection connection) throws Exception {
+      try {
 
-        server.handleConnection(connection);
-
-      }
+        LOG.info("Established connection...");
+        final Connection connection = serverSource.connect();
+        LOG.info("Connection established.");
 
-      @Override
-      public void onException(Exception ex) {
-
-        System.out.println("ERROR: Exception during server tick. Check log for details.");
-        LOG.error(ex, "Exception during server tick.");
+        server.handleConnection(connection);
 
+      } catch (IOException ex) {
+        LOG.error(ex, "Failed to establish connection.");
       }
-    });
-
-    LOG.info("Starting hub...");
-
-    hub.run();
-
-    LOG.info("Hub exited.");
+    }
   }
 }

src/codeu/chat/common/Hub.java

@@ -35,12 +35,15 @@
 import codeu.chat.common.Uuids;
 import codeu.chat.util.Logger;
 import codeu.chat.util.Serializers;
+import codeu.chat.util.Timeline;
 import codeu.chat.util.connections.Connection;
 
 public final class Server {
 
   private final static Logger.Log LOG = Logger.newLog(Server.class);
 
+  private final Timeline timeline = new Timeline();
+
   private final Uuid id;
   private final byte[] secret;
 
@@ -58,20 +61,57 @@ public Server(Uuid id, byte[] secret, Relay relay) {
 
     this.controller = new Controller(id, model);
     this.relay = relay;
-  }
 
-  public void syncWithRelay(int maxReadSize) throws Exception {
-    for (final Relay.Bundle bundle :  relay.read(id, secret, lastSeen, maxReadSize)) {
-      onBundle(bundle);
-      lastSeen = bundle.id();
-    }
+    timeline.scheduleNow(new Runnable() {
+      @Override
+      public void run() {
+        try {
+
+          LOG.info("Reading update from relay...");
+
+          for (final Relay.Bundle bundle : relay.read(id, secret, lastSeen, 32)) {
+            onBundle(bundle);
+            lastSeen = bundle.id();
+          }
+
+        } catch (Exception ex) {
+
+          LOG.error(ex, "Failed to read update from relay.");
+
+        }
+
+        // Do this all again in 60 seconds
+        timeline.scheduleIn(60000, this);
+      }
+    });
   }
 
-  public boolean handleConnection(Connection connection) throws Exception {
+  public void handleConnection(final Connection connection) {
+    timeline.scheduleNow(new Runnable() {
+      @Override
+      public void run() {
+        try {
+
+          LOG.info("Handling connection...");
+
+          final boolean success = onMessage(
+              connection.in(),
+              connection.out());
+
+          LOG.info("Connection handled: %s", success ? "ACCEPTED" : "REJECTED");
+        } catch (Exception ex) {
 
-    LOG.info("Handling new connection...");
+          LOG.error(ex, "Exception while handling connection.");
 
-    return onMessage(connection.in(), connection.out());
+        }
+
+        try {
+          connection.close();
+        } catch (Exception ex) {
+          LOG.error(ex, "Exception while closing connection.");
+        }
+      }
+    });
   }
 
   private boolean onMessage(InputStream in, OutputStream out) throws IOException {
@@ -89,13 +129,10 @@ private boolean onMessage(InputStream in, OutputStream out) throws IOException {
       Serializers.INTEGER.write(out, NetworkCode.NEW_MESSAGE_RESPONSE);
       Serializers.nullable(Message.SERIALIZER).write(out, message);
 
-      // Unlike the other calls - we need to send something the result of this
-      // call to the relay. Waiting until after the server has written back to
-      // the client allows the client to get the response, but the network
-      // connection has not been closed. However to wait after until the client-server
-      // connection was closed before sending would be very complicated.
-
-      sendToRelay(author, conversation, message.id);
+      timeline.scheduleNow(createSendToRelayEvent(
+          author,
+          conversation,
+          message.id));
 
     } else if (type == NetworkCode.NEW_USER_REQUEST) {
 
@@ -252,16 +289,21 @@ private void onBundle(Relay.Bundle bundle) {
     }
   }
 
-  private void sendToRelay(Uuid userId, Uuid conversationId, Uuid messageId) {
-
-    final User user = view.findUser(userId);
-    final Conversation conversation = view.findConversation(conversationId);
-    final Message message = view.findMessage(messageId);
-
-    relay.write(id,
-                secret,
-                relay.pack(user.id, user.name, user.creation),
-                relay.pack(conversation.id, conversation.title, conversation.creation),
-                relay.pack(message.id, message.content, message.creation));
+  private Runnable createSendToRelayEvent(final Uuid userId,
+                                          final Uuid conversationId,
+                                          final Uuid messageId) {
+    return new Runnable() {
+      @Override
+      public void run() {
+        final User user = view.findUser(userId);
+        final Conversation conversation = view.findConversation(conversationId);
+        final Message message = view.findMessage(messageId);
+        relay.write(id,
+                    secret,
+                    relay.pack(user.id, user.name, user.creation),
+                    relay.pack(conversation.id, conversation.title, conversation.creation),
+                    relay.pack(message.id, message.content, message.creation));
+      }
+    };
   }
 }