Merge branch 'live_update'

bwitt · Oct 9, 2015 · 1bc82b2 · 1bc82b2
2 parents 2e3befb + d53c34f
commit 1bc82b2
Show file tree

Hide file tree

Showing 12 changed files with 459 additions and 200 deletions.
diff --git a/code/main.rb b/code/main.rb
@@ -145,7 +145,7 @@ def flash(type, msg)
     end
 
     get '/' do
-      erb :main
+      erb :main, :locals => { :main_path => "#{root}/q" }
     end
 
     get '/new_queue' do
@@ -200,30 +200,42 @@ def flash(type, msg)
       content_type 'application/json'
       builtin_queues, custom_queues = queuemgr.queues.sort.partition { |q| builtin_queue? q }
 
-      (custom_queues.map do |name|
-        qc = get_queueclient(name)
-        {
-          'name'   => name,
-          'status' => qc.status,
-          'ping'   => qc.ping,
-          'pid'    => qc.read_pid,
-          'uptime' => qc.uptime,
-          'counts' => Hash[ msgs_labels.zip(qc.num_messages.values_at(*msgs_labels)) ],
-          # 'schedule' => qc.config[1]['schedule']...
-        }
-      end +
-      builtin_queues.map do |name|
-        qc = get_queueclient(name)
-        {
-          'name'   => name,
-          'status' => qc.status,
-          'ping'   => qc.ping,
-          'pid'    => qc.read_pid,
-          'uptime' => qc.uptime,
-          'counts' => Hash[ msgs_labels.zip(qc.num_messages.values_at(*msgs_labels)) ],
-          # 'schedule' => qc.config[1]['schedule']...
-        }
-      end).to_json
+      {
+        'status'  => queuemgr.running? ? 'OPERATIONAL' : 'DOWN',
+        'ping'    => queuemgr.ping,
+        'pid'     => queuemgr.read_pid,
+        'uptime'  => queuemgr.uptime,
+        'version' => queuemgr.version,
+        'time'    => Time.now.to_i,
+        'queues'  => (
+          custom_queues.map do |name|
+            qc = get_queueclient(name)
+            counts = Hash[ msgs_labels.zip(qc.num_messages.values_at(*msgs_labels)) ] rescue {}
+            {
+              'name'   => name,
+              'status' => qc.status,
+              'ping'   => qc.ping,
+              'pid'    => qc.read_pid,
+              'uptime' => qc.uptime,
+              'counts' => counts,
+              # 'schedule' => qc.config[1]['schedule']...
+            }
+          end +
+          builtin_queues.map do |name|
+            qc = get_queueclient(name)
+            counts = Hash[ msgs_labels.zip(qc.num_messages.values_at(*msgs_labels)) ] rescue {}
+            {
+              'name'   => name,
+              'status' => qc.status,
+              'ping'   => qc.ping,
+              'pid'    => qc.read_pid,
+              'uptime' => qc.uptime,
+              'counts' => counts,
+              # 'schedule' => qc.config[1]['schedule']...
+            }
+          end
+        ),
+      }.to_json
     end
 
     get '/search' do
@@ -278,24 +290,28 @@ def flash(type, msg)
         nm = qc.num_messages
         return {
           'status'       => qc.status,
+          'ping'         => qc.ping,
+          'pid'          => qc.read_pid,
           'uptime'       => qc.uptime,
+          'time'         => Time.now.to_i,
           'prep_size'    => nm['prep'],
           'que_size'     => nm['que'],
           'run_size'     => nm['run'],
           'done_size'    => nm['done'],
           'err_size'     => nm['err'],
           'relayed_size' => nm['relayed'],
-          'messages' => {
-            'prep' => qc.messages({'state' => 'prep'}),
-            'que'  => qc.messages({'state' => 'que'}),
-            'run'  => qc.messages({'state' => 'run'}),
-            'done' => qc.messages({'state' => 'done'}),
-            'err'  => qc.messages({'state' => 'err'}),
+          'messages'  => {
+            'prep'    => qc.messages({'state' => 'prep'}),
+            'que'     => qc.messages({'state' => 'que'}),
+            'run'     => qc.messages({'state' => 'run'}),
+            'done'    => qc.messages({'state' => 'done'}),
+            'err'     => qc.messages({'state' => 'err'}),
+            'relayed' => qc.messages({'state' => 'relayed'}),
           }
         }.to_json
       else
         ok, config = qc.config
-        erb :queue, :locals => { :qc => qc, :config => config }
+        erb :queue, :locals => { :qc => qc, :config => config, :queue_path => "#{root}q/#{params[:name]}" }
       end
     end
 
@@ -789,6 +805,18 @@ def flash(type, msg)
           end
           redirect "#{root}q/#{params[:name]}/#{params[:msg_id]}"
         end
+      when 'run_now'
+        res = qc.run_message({ 'msg_id' => params[:msg_id] })
+
+        if not res
+          throw :halt, [500, "500 - Couldn't run message. Internal error."]
+        end
+        if res[0] != 'ok'
+          throw :halt, [500, "500 - Couldn't run message. #{res.inspect}."]
+        end
+
+        flash :notice, "Message in run successfully"
+        redirect "#{root}q/#{params[:name]}/#{params[:msg_id]}"
       else
         throw :halt, [400, "400 - Invalid method param"]
       end

diff --git a/code/public/css/rq.css b/code/public/css/rq.css
@@ -63,10 +63,25 @@ td {
   background: #FEF3DD;
 }
 
-.meta {
+hr.meta {
+  background: #AB9B7D;
+  color: #FEF9EE;
+  height: 10px;
+  margin-top: 10px;
+}
+
+table#meta {
   margin-top: 20px;
 }
 
+#meta td, #meta th {
+  padding: 5px;
+  text-align: left;
+  background: #AB9B7D;
+  color: #FEF9EE;
+  white-space: nowrap;
+}
+
 #env {
   font-weight: bold;
 }
@@ -98,6 +113,22 @@ td {
   color: black;
 }
 
+.template-dest:before {
+  content: "relay dest: ";
+}
+
+.template-status:before {
+  content: "status: ";
+}
+
+.template-seconds:before {
+  content: "in ";
+}
+
+.template-seconds:after {
+  content: " secs";
+}
+
 ul {
   list-style: none;
 }
@@ -168,7 +199,9 @@ h1 {
 
 #queue-status, #message-actions {
   float: right;
-  text-align: right;
+  position: relative;
+  bottom: -4px;
+  right: 0px;
 }
 
 .right {
@@ -231,6 +264,7 @@ a:visited {
 
 pre {
   margin: 0;
+  display: inline;
 }
 
 #filedata {
@@ -336,10 +370,13 @@ form.inrow button {
 #queue-lists h5 {
   font-size: 1.5em;
   margin-bottom: 0;
+  padding-top: 15px;
 }
 
 #queue-lists ul {
   padding: 0;
+  margin: 0px;
+  margin-top: -15px;
 }
 
 #queue-lists li {
@@ -375,15 +412,15 @@ th.done {
   padding: 10px;
 }
 
-.relayed-li {
+.relayed li {
   margin-bottom: 15px;
 }
 
-.relayed-li a {
+.relayed li a {
   display: inline;
 }
 
-.relayed-li span {
+.relayed li span {
   padding: 5px;
 }
 

diff --git a/code/public/js/main.js b/code/public/js/main.js
@@ -0,0 +1,56 @@
+domready(function() {
+  // Update the page and all queue rows
+  function displayQueues(queues) {
+    document.getElementById("status").innerHTML = queues["status"];
+    document.getElementById("ping").innerHTML = queues["ping"];
+    document.getElementById("pid").innerHTML = queues["pid"];
+    document.getElementById("uptime").innerHTML = queues["uptime"];
+    document.getElementById("version").innerHTML = queues["version"];
+    document.getElementById("time").innerHTML = queues["time"];
+
+    for (var row of queues["queues"]) {
+      document.getElementById("ping-" + row["name"]).innerHTML = row["ping"];
+      document.getElementById("pid-" + row["name"]).innerHTML = row["pid"];
+      document.getElementById("uptime-" + row["name"]).innerHTML = row["uptime"];
+
+      var s = document.getElementById("status-" + row["name"]);
+      if (row["status"] == "UP") {
+        s.innerHTML = "UP";
+        s.className = "green";
+      } else {
+        s.innerHTML = row["status"];
+        s.className = "red";
+      }
+
+      for (var c in row["counts"]) {
+        var len = row["counts"][c].toString().length;
+        var justy = row["counts"][c] + " ".repeat(len < 4 ? 4 - len : 0);
+        document.getElementById(c + "-" + row["name"]).innerHTML = justy;
+      }
+    }
+  }
+
+  // Fetch q.json
+  function nextUpdate() {
+    var xmlhttp = new XMLHttpRequest();
+    xmlhttp.open("GET", config.main_path + ".json", true);
+    xmlhttp.setRequestHeader("Cache-Control", "no-cache");
+    xmlhttp.onreadystatechange = function() {
+      // State 4 means "totally done"
+      if (xmlhttp.readyState == 4) {
+        // Schedule the next fetch
+        window.setTimeout(function () { nextUpdate(); }, 900);
+
+        // Process the result
+        if (xmlhttp.status == 200) {
+          var responseJSON = JSON.parse(xmlhttp.responseText);
+          displayQueues(responseJSON);
+        }
+      }
+    };
+    xmlhttp.send();
+  }
+
+  // Schedule the first fetch
+  nextUpdate();
+});