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/RelayMain.java

@@ -16,10 +16,10 @@
 
 import java.io.IOException;
 
-import codeu.chat.common.Hub;
 import codeu.chat.relay.Server;
 import codeu.chat.relay.ServerFrontEnd;
 import codeu.chat.util.Logger;
+import codeu.chat.util.Timeline;
 import codeu.chat.util.connections.Connection;
 import codeu.chat.util.connections.ConnectionSource;
 import codeu.chat.util.connections.ServerConnectionSource;
@@ -56,39 +56,39 @@ public static void main(String[] args) {
   private static void startRelay(ConnectionSource source) {
 
     final Server relay = new Server(1024, 16);
-
     LOG.info("Relay object created.");
 
-    // TODO: Load team information
-
     final ServerFrontEnd frontEnd = new ServerFrontEnd(relay);
-
     LOG.info("Relay front end object created.");
 
-    LOG.info("Starting relay main loop...");
-
-    final Runnable hub = new Hub(source, new Hub.Handler() {
-
-      @Override
-      public void handle(Connection connection) throws Exception {
-
-        frontEnd.handleConnection(connection);
+    final Timeline timeline = new Timeline();
+    LOG.info("Relay timeline created.");
 
-      }
-
-      @Override
-      public void onException(Exception ex) {
+    // TODO: Load team information
 
-        System.out.println("ERROR: front end failed to handle connection. Check log for details.");
-        LOG.error(ex, "Exception handling connection.");
+    LOG.info("Starting relay main loop...");
 
+    while (true) {
+      try {
+
+        LOG.info("Established connection...");
+        final Connection connection = source.connect();
+        LOG.info("Connection established.");
+
+        timeline.scheduleNow(new Runnable() {
+          @Override
+          public void run() {
+            try {
+              frontEnd.handleConnection(connection);
+            } catch (Exception ex) {
+              LOG.error(ex, "Exception handling connection.");
+            }
+          }
+        });
+
+      } catch (IOException ex) {
+        LOG.error(ex, "Failed to establish connection.");
       }
-    });
-
-    LOG.info("Starting hub...");
-
-    hub.run();
-
-    LOG.info("Hub exited.");
+    }
   }
 }

src/codeu/chat/ServerMain.java

@@ -17,7 +17,6 @@
 
 import java.io.IOException;
 
-import codeu.chat.common.Hub;
 import codeu.chat.common.Relay;
 import codeu.chat.common.Secret;
 import codeu.chat.common.Uuid;

src/codeu/chat/server/Server.java

@@ -80,8 +80,8 @@ public void run() {
 
         }
 
-        // Do this all again in 60 seconds
-        timeline.scheduleIn(60000, this);
+        // Do this again in 5 seconds
+        timeline.scheduleIn(5000, this);
       }
     });
   }

src/codeu/chat/util/Timeline.java

@@ -26,6 +26,8 @@
 // code has been executed.
 public final class Timeline {
 
+  private final static Logger.Log LOG = Logger.newLog(Timeline.class);
+
   private static final class Event implements Comparable<Event> {
 
     public final long time;
@@ -49,29 +51,55 @@ public int compareTo(Event other) {
 
   // This thread is used to track the time of events and moves events from the
   // "backlog" queue to the "todo" queue when it is time to execute. They are
-  // seperated to allow this thread to be safely interrupted.
+  // separated to allow this thread to be safely interrupted.
   private final Thread scheduler = new Thread() {
     @Override
     public void run() {
       while (running) {
-        final long now = System.currentTimeMillis();
-        final Event next = backlog.poll();
+
+        Event next;
+
         try {
-          if (next == null) {
-            Thread.sleep(10000);
-          } else if (next.time > now) {
-            // Put it back (it's not time).
-            while (!backlog.offer(next)) {
-              // force this to go through
-            }
-            Thread.sleep(next.time - now);
+          next = backlog.take();
+        } catch (InterruptedException ex) {
+          // Rather than try to handle the exception here, set "next"
+          // to null and let the normal flow handle the case.
+          next = null;
+        }
+
+        long sleep = 0;
+
+        if (next != null) {
+
+          final long now = System.currentTimeMillis();
+
+          // Check which queue the event should be added to. If it
+          // is time to execute, it should be added to the "todo"
+          // queue. If it is not time, it should be added back to the
+          // "backlog".
+          // If the item is added back to the backlog, we know how long
+          // it will be until it will be executed. That means we can sleep
+          // until then.
+          if (next.time <= now) {
+            forceAdd(todo, next.callback);
+            sleep = 0;
           } else {
-            while (!todo.offer(next.callback)) {
-              // force this to go through
-            }
+            // Put it back (it's not time).
+            forceAdd(backlog, next);
+            sleep = next.time - now;
+          }
+        }
+
+        if (sleep > 0) {
+          try {
+            Thread.sleep(sleep);
+          } catch (InterruptedException ex) {
+            // There are two cases this will happen:
+            //  1. A new item was added and we are being woken to
+            //     check if we need to update the time.
+            //  2. It is time to exit and we need to wake-up so that
+            //     we can check that "running" is "false".
           }
-        } catch (InterruptedException ex) {
-          // A new event was added - need to recheck all the times.
         }
       }
     }
@@ -87,8 +115,11 @@ public void run() {
         try {
           todo.take().run();
         } catch (Exception ex) {
-          // Catch all exceptions here to stop any rouge action from
+          // Catch all exceptions here to stop any rogue action from
           // take down the timeline.
+          LOG.warning(
+              "An exception was seen on the timeline (%s)",
+              ex.toString());
         }
       }
     }
@@ -120,27 +151,45 @@ public void scheduleIn(long ms, Runnable callback) {
   // point in time.
   public void scheduleAt(long timeMs, Runnable callback) {
     final Event event = new Event(timeMs, callback);
-    while (!backlog.offer(event)) {
-      // force add
-    }
+    forceAdd(backlog, event);
     scheduler.interrupt();  // wake it up
   }
 
+  // STOP
+  //
+  // Tell the timeline to shutdown. This is a non-blocking call.
   public void stop() {
     running = false;
-    forceStop(executor);
-    forceStop(scheduler);
+
+    // Interrupt does not force a thread to exit. It signal's the
+    // thead that it is time to stop execution. As the threads may
+    // be sleeping, this will force them awake.
+    executor.interrupt();
+    scheduler.interrupt();
   }
 
-  private static void forceStop(Thread thread) {
+  // JOIN
+  //
+  // Wait for the timeline to shutdown. This is a blocking call.
+  public void join() {
+    forceJoin(executor);
+    forceJoin(scheduler);
+  }
+
+  private static void forceJoin(Thread thread) {
     while (true) {
       try {
-        thread.interrupt();
         thread.join();
         break;
       } catch (InterruptedException ex) {
         // Do nothing - allow this to try again.
       }
     }
   }
+
+  private static <T> void forceAdd(BlockingQueue<T> queue, T value) {
+    while (!queue.offer(value)) {
+      // try again...
+    }
+  }
 }