Replace monitor comet socket with a websocket

app/controllers/Monitor.scala

@@ -3,6 +3,7 @@ package controllers
 import play.api.mvc._
 import play.api.libs.Comet
 import play.api.libs.concurrent._
+import play.api.libs.json._
 import akka.pattern.ask
 import akka.util.duration._
 import akka.util.Timeout
@@ -17,11 +18,11 @@ object Monitor extends LilaController {
   implicit val timeout = Timeout(100 millis)
 
   val index = Action {
-    Ok(views.html.monitor.monitor(env.monitor.stream.maxMemory))
+    Ok(views.html.monitor.monitor(monitor.Reporting.maxMemory))
   }
 
-  val stream = Action {
-    Ok.stream(env.monitor.stream.getData &> Comet(callback = "parent.message"))
+  val websocket = WebSocket.async[JsValue] { implicit req ⇒
+    env.monitor.socket.join(uidOption = get("uid", req))
   }
 
   val status = Action {

app/core/Settings.scala

@@ -77,6 +77,7 @@ final class Settings(config: Config) {
   val ActorSiteHub = "site_hub"
   val ActorGameHubMaster = "game_hub_master"
   val ActorLobbyHub = "lobby_hub"
+  val ActorMonitorHub = "monitor_hub"
 
   private def millis(name: String): Int = getMilliseconds(name).toInt
 

app/monitor/Hub.scala

@@ -0,0 +1,22 @@
+package lila
+package monitor
+
+import socket._
+
+import akka.actor._
+import play.api.libs.json._
+import play.api.libs.iteratee._
+
+final class Hub(timeout: Int) extends HubActor[Member](timeout) {
+
+  def receiveSpecific = {
+
+    case Join(uid) ⇒ {
+      val channel = new LilaEnumerator[JsValue](Nil)
+      addMember(uid, Member(channel))
+      sender ! Connected(channel)
+    }
+
+    case MonitorData(data) ⇒ notifyAll("monitor", JsString(data mkString ";"))
+  }
+}

app/monitor/MonitorEnv.scala

@@ -16,16 +16,18 @@ final class MonitorEnv(
   implicit val ctx = app
   import settings._
 
+  lazy val hub = Akka.system.actorOf(
+    Props(new Hub(timeout = SiteUidTimeout)), name = ActorMonitorHub)
+
+  lazy val socket = new Socket(hub = hub)
+
   lazy val reporting = Akka.system.actorOf(
     Props(new Reporting(
       rpsProvider = rpsProvider,
-      mongodb = mongodb
+      mongodb = mongodb,
+      hub = hub
     )), name = ActorReporting)
 
   val rpsProvider = new RpsProvider(
     timeout = MonitorTimeout)
-
-  val stream = new Stream(
-    reporting = reporting,
-    timeout = MonitorTimeout)
 }

app/monitor/Reporting.scala

@@ -17,7 +17,8 @@ import com.mongodb.casbah.MongoDB
 
 final class Reporting(
   rpsProvider: RpsProvider,
-  mongodb: MongoDB
+  mongodb: MongoDB,
+  hub: ActorRef
 ) extends Actor {
 
   case class SiteSocket(nbMembers: Int)
@@ -93,7 +94,10 @@ final class Reporting(
         }
       } onComplete {
         case Left(a) ⇒ println("Reporting: " + a.getMessage)
-        case a       ⇒ display()
+        case a       ⇒ {
+          hub ! MonitorData(monitorData)
+          display()
+        }
       }
     }
   }
@@ -165,3 +169,8 @@ final class Reporting(
     }
   }
 }
+
+object Reporting {
+
+  def maxMemory = Runtime.getRuntime().totalMemory() / (1024 * 1024)
+}

app/monitor/Socket.scala

@@ -0,0 +1,38 @@
+package lila
+package monitor
+
+import akka.actor._
+import akka.pattern.ask
+import akka.util.duration._
+import akka.util.Timeout
+import play.api.libs.json._
+import play.api.libs.iteratee._
+import play.api.libs.concurrent._
+import scalaz.effects._
+
+import implicits.RichJs._
+import socket.{ Util, Ping, Quit }
+
+final class Socket(hub: ActorRef) {
+
+  implicit val timeout = Timeout(300 millis)
+
+  def join(
+    uidOption: Option[String]): SocketPromise = {
+    val promise: Option[SocketPromise] = for {
+      uid ← uidOption
+    } yield (hub ? Join(uid)).asPromise map {
+      case Connected(channel) ⇒
+        val iteratee = Iteratee.foreach[JsValue] { e ⇒
+          e str "t" match {
+            case Some("p") ⇒ hub ! Ping(uid)
+            case _         ⇒
+          }
+        } mapDone { _ ⇒
+          hub ! Quit(uid)
+        }
+        (iteratee, channel)
+    }
+    promise | Util.connectionFail
+  }
+}

app/monitor/Stream.scala

@@ -1,29 +0,0 @@
-package lila
-package monitor
-
-import play.api.libs.iteratee._
-import play.api.libs.concurrent.Promise
-import akka.actor._
-import akka.dispatch.Await
-import akka.pattern.ask
-import akka.util.duration._
-import akka.util.Timeout
-
-final class Stream(
-    reporting: ActorRef,
-    timeout: Int) {
-
-  implicit val maxWait = 100 millis
-  implicit val maxWaitTimeout = Timeout(maxWait)
-
-  val getData = Enumerator.generateM {
-    Promise.timeout(Some(data mkString ";"), timeout)
-  }
-
-  def maxMemory = Runtime.getRuntime().totalMemory() / (1024 * 1024)
-
-  private def data = Await.result(
-    reporting ? GetMonitorData mapTo manifest[List[String]],
-    maxWait
-  ) 
-}

app/monitor/actorApi.scala

@@ -2,6 +2,7 @@ package lila
 package monitor
 
 import core.CoreEnv
+import socket.SocketMember
 
 case object GetNbGames
 case object GetNbPlaying
@@ -9,3 +10,11 @@ case object GetStatus
 case object GetMonitorData
 
 case class Update(env: CoreEnv)
+
+case class Member(channel: Channel) extends SocketMember {
+  val username = none
+}
+
+case class Join(uid: String)
+case class Connected(channel: Channel)
+case class MonitorData(data: List[String])

app/views/monitor/layout.scala.html

@@ -0,0 +1,22 @@
+@(title: String)(content: Html)
+<!DOCTYPE html>
+<html>
+  <head>
+    <title>@title</title>
+    <meta name="viewport" content="width=device-width, user-scalable=no" />
+    <meta name="apple-mobile-web-app-capable" content="yes" />
+    <meta name="apple-mobile-web-app-status-bar-style" content="black" />
+    <link rel="stylesheet" media="screen" href="@routes.Assets.at("stylesheets/monitor.css")" />
+    <link rel="apple-touch-icon-precomposed" href="@routes.Assets.at("images/icon.png")"/>
+    <link rel="apple-touch-icon-precomposed" sizes="72x72" href="@routes.Assets.at("images/icon.png")">
+    <link rel="apple-touch-icon-precomposed" sizes="114x114" href="@routes.Assets.at("images/icon@2x.png")">
+    <link rel="shortcut icon" type="image/png" href="@routes.Assets.at("images/favicon.png")" />
+    <script src="@routes.Assets.at("vendor/zanimo.min.js")" type="text/javascript"></script>
+  </head>
+  <body>
+    @content
+  </body>
+  @jsTag("deps.min.js")
+  @jsTag("socket.js")
+  @jsTag("ctrl.js")
+</html>

app/views/monitor/main.scala.html

@@ -1,19 +0,0 @@
-@(title: String)(content: Html)
-<!DOCTYPE html>
-<html>
-    <head>
-        <title>@title</title>
-        <meta name="viewport" content="width=device-width, user-scalable=no" />
-        <meta name="apple-mobile-web-app-capable" content="yes" />
-        <meta name="apple-mobile-web-app-status-bar-style" content="black" />
-        <link rel="stylesheet" media="screen" href="@routes.Assets.at("stylesheets/monitor.css")" />
-        <link rel="apple-touch-icon-precomposed" href="@routes.Assets.at("images/icon.png")"/>
-        <link rel="apple-touch-icon-precomposed" sizes="72x72" href="@routes.Assets.at("images/icon.png")">
-        <link rel="apple-touch-icon-precomposed" sizes="114x114" href="@routes.Assets.at("images/icon@2x.png")">
-        <link rel="shortcut icon" type="image/png" href="@routes.Assets.at("images/favicon.png")" />
-        <script src="@routes.Assets.at("vendor/zanimo.min.js")" type="text/javascript"></script>
-    </head>
-    <body>
-        @content
-    </body>
-</html>

app/views/monitor/monitor.scala.html

@@ -1,17 +1,17 @@
 @(totalMemory: Long)
 
-@main("RPS") {
+@layout("RPS") {
 <h1>Lichess reactor <span class="down">DOWN</span><span class="up">OPERATIONAL</span></h1>
 <div id="monitors" class="clearfix">
   <script type="text/javascript">window.App = { }; window.App.totalMemory = @totalMemory;</script>
   <script type="text/javascript" src="@routes.Assets.at("javascripts/monitor.js")"></script>
 </div>
 <div id="actions">
-<a href="#" id="shutdown">Emergency reactor shutdown - DON'T CLICK</a>
+  <a href="#" id="shutdown">Emergency reactor shutdown - DON'T CLICK</a>
 </div>
 <script>
-window.document.getElementById("shutdown").onclick = function() {
+  window.document.getElementById("shutdown").onclick = function() {
   alert("Was worth trying. I guess.");
-};
+  };
 </script>
 }

conf/routes

@@ -111,7 +111,7 @@ POST  /inbox/$id<[\w]{8}>/delete       controllers.Message.delete(id: String)
 
 # Monitor
 GET   /monitor                         controllers.Monitor.index
-GET   /monitor/stream                  controllers.Monitor.stream
+GET   /monitor/socket                  controllers.Monitor.websocket
 GET   /nb-players                      controllers.Monitor.nbPlayers
 GET   /nb-playing                      controllers.Monitor.nbPlaying
 GET   /status                          controllers.Monitor.status

public/javascripts/ctrl.js

@@ -11,7 +11,9 @@ var lichess = {
     events: {
       n: function(e) {
         var $tag = $('#nb_connected_players');
-        $tag.html($tag.html().replace(/\d+/, e)).removeClass('none');
+        if ($tag.length) {
+          $tag.html($tag.html().replace(/\d+/, e)).removeClass('none');
+        }
       },
       nbm: function(e) {
         var $tag = $('#nb_messages');

public/javascripts/monitor.js

@@ -156,34 +156,35 @@
       container : container
     });
 
-    var iframe = create("iframe");
-    iframe.src = "/monitor/stream";
-    iframe.style.display = "none";
-
-    window.message = function (msg) {
-      console.debug(msg);
-      var ds = msg.split(";");
-      app.lastCall = (new Date()).getTime();
-      for(var i in ds) {
-        var d = ds[i].split(":");
-        if (d.length == 2) {
-          if (typeof app[d[1]] != "undefined") {
-            app[d[1]].update(d[0]);
-          }
-        }
-      }
+    function setStatus(s) {
+      window.document.body.className = s;
     }
 
-    setTimeout(function () {
-      app.lastCall = (new Date()).getTime();
-      window.document.body.appendChild(iframe);
-    }, 100);
+    lichess.socket = new $.websocket(lichess.socketUrl + "/monitor/socket", 0, $.extend(true, lichess.socketDefaults, {
+      events: {
+        monitor: function(msg) {
+          var ds = msg.split(";");
+          app.lastCall = (new Date()).getTime();
+          for(var i in ds) {
+            var d = ds[i].split(":");
+            if (d.length == 2) {
+              if (typeof app[d[1]] != "undefined") {
+                app[d[1]].update(d[0]);
+              }
+            }
+          }
+        }
+      },
+      options: {
+        name: "monitor"
+      }
+    }));
 
     setInterval(function () {
       if ((new Date()).getTime() - app.lastCall > 3000) {
-        window.document.body.className = "down";
-      } else {
-        window.document.body.className = "up";
+        setStatus("down");
+      } else if (app.lastCall) {
+        setStatus("up");
       }
     },1100);
   }

public/javascripts/socket.js

@@ -58,10 +58,8 @@ $.websocket.prototype = {
       var m = JSON.parse(e.originalEvent.data);
       if (m.t == "n") {
         self.keepAlive();
-        self._debug(m);
-      } else {
-        self._debug(m);
       }
+        self._debug(m);
       if (m.t == "batch") {
         $(m.d || []).each(function() { self._handle(this); });
       } else {

public/stylesheets/monitor.css

@@ -24,6 +24,7 @@ h1 .down {
 }
 h1 .up {
   color: #446644;
+  display: none;
 }
 body.down h1 .down {
   display: inline;