Support of unknown number of feeders + display log messages

server.py

@@ -1,36 +1,52 @@
 #!/usr/bin/env python3.5
 from flask import Flask, render_template, Response
 import json
+import redis
+import configparser
 from time import gmtime as now
 from time import sleep, strftime
+import os
+
+configfile = os.path.join(os.environ['VIRTUAL_ENV'], '../config.cfg')
+cfg = configparser.ConfigParser()
+cfg.read(configfile)
 
 app = Flask(__name__)
 
-FIELDNAME = {
+redis_server = redis.StrictRedis(
-        'time': 'Time', 
+        host=cfg.get('Redis', 'host'),
-        'level': 'Level', 
+        port=cfg.getint('Redis', 'port'),
-        'source': 'Source', 
+        db=cfg.getint('Redis', 'db'))
-        'name': 'Name', 
-        'message': 'Message'
-        }
 
-FIELDNAME_ORDER = [ 'time', 'level', 'source', 'name', 'message' ]
+subscriber = redis_server.pubsub(ignore_subscribe_messages=True)
+subscriber.psubscribe(cfg.get('Redis', 'channel'))
+eventNumber = 0
 
-class LogRow():
+class LogItem():
-    def __init__(self, feed='', time='', level='level', src='source', name='name', message='wonderful meesage'):
+
+    FIELDNAME_ORDER = [ 'time', 'level', 'source', 'name', 'message' ]
+
+    #def __init__(self, feed='', time='', level='level', src='source', name='name', message='wonderful meesage'):
+    def __init__(self, feed):
         # Parse feed message
         
-        # Assign potential supplied values
+        ## Assign potential supplied values
-        self.time = time if time != '' else strftime("%H:%M:%S", now())
+        #self.time = time if time != '' else strftime("%H:%M:%S", now())
-        self.level = level
+        #self.level = level
-        self.source = src
+        #self.source = src
-        self.name = name
+        #self.name = name
-        self.message = message
+        #self.message = message
+
+        self.time = strftime("%H:%M:%S", now())
+        self.level = 'level'
+        self.source = 'src'
+        self.name = 'name'
+        self.message = feed
 
     def get_head_row(self):
         to_ret = []
-        for fn in FIELDNAME_ORDER:
+        for fn in LogItem.FIELDNAME_ORDER:
-            to_ret.append(FIELDNAME[fn])
+            to_ret.append(fn[0].upper()+fn[1:])
         return to_ret
 
     def get_row(self):
@@ -45,26 +61,32 @@ class LogRow():
 
 
 class EventMessage():
+    # Suppose the event message is a json with the format {name: 'feedName', log:'logData'}
     def __init__(self, msg):
-        self.feed = None
+        msg = msg.decode('utf8')
-        self.isLog = EventMessage.is_log(msg)
+        try:
+            jsonMsg = json.loads(msg)
+        except json.JSONDecodeError:
+            jsonMsg = { 'name': "undefined" ,'log': msg }
+
+        self.feedName = jsonMsg['name']
+        self.feed = jsonMsg['log']
+        self.feed = LogItem(jsonMsg['log']).get_row()
 
         #get type of message: log or feed, then create
-        if self.isLog:
+        #if self.isLog:
-            self.feed = msg
+        #    self.feed = msg
-            #FIXME do parser
+        #    #FIXME do parser
-            self.feed = LogRow(feed=msg).get_row()
+        #    self.feed = LogItem(feed=msg).get_row()
-        else:
+        #else:
-            self.feed = {feed.name: feed.data}
+        #    #FIXME do parser
-
+        #    temp = []
-    def is_log(msg):
+        #    for feed in msg:
-        return True
+        #        temp.append(feed.name, feed.data)
+        #    self.feed = temp
 
     def to_json(self):
-        if self.isLog:
+        to_ret = { 'log': self.feed, 'feedName': self.feedName }
-            to_ret = { 'log': self.feed, 'chart': "" }
-        else:
-            to_ret = { 'log': "", 'chart': self.feed }
         return 'data: {}\n\n'.format(json.dumps(to_ret))
 
 @app.route("/")
@@ -77,16 +99,12 @@ def logs():
 
 @app.route("/_get_log_head")
 def getLogHead():
-    return json.dumps(LogRow().get_head_row())
+    return json.dumps(LogItem('').get_head_row())
 
 def event_stream():
-    #for msg in pubsub:
+    for msg in subscriber.listen():
-    for i in range(3):
+        content = msg['data']
-        msg = now()
+        yield EventMessage(content).to_json()
-        sleep(0.3)
-        print('sending', EventMessage(msg).to_json())
-        yield EventMessage(msg).to_json()
-
 
 if __name__ == '__main__':
-    app.run(host='localhost', port=8000, debug=True)
+    app.run(host='localhost', port=8000)

static/js/index.js

@@ -1,19 +1,104 @@
+//color: #5f6062
+var updateInterval = 1000; // 1s
+var numPoint = 60;
 
-var updateInterval = 30; // 30ms
-var numPoint = 10;
 var emptyArray = [];
+var dataNumLog = [];
 for(i=0; i<numPoint; i++) {
-    emptyArray.push(0);
+    dataNumLog.push([i, 0]);
+    emptyArray.push([i, 0]);
 }
-var data = { 'undefined': { label: 'undefined', data: [] } };
+
-var curMax = { 'undefined': [] };
+class Sources {
+    constructor() {
+        this._sourcesArray = {};
+        this._sourcesCount = {};
+        this._sourcesCountMax = {};
+        this._globalMax = 0;
+        this._sourceNames = [];
+    }
+
+    addSource(sourceName) {
+        this._sourcesArray[sourceName] = emptyArray;
+        this._sourcesCount[sourceName] = 0;
+        this._sourcesCountMax[sourceName] = 0;
+        this._sourceNames.push(sourceName);
+    }
+
+    addIfNotPresent(sourceName) {
+        if (this._sourceNames.indexOf(sourceName) == -1) {
+            this.addSource(sourceName);
+        }
+    }
+
+    incCountOnSource(sourceName) {
+        this._sourcesCount[sourceName] += 1;
+    }
+
+    resetCountOnSource() {
+        for (var src of this._sourceNames) {
+            this._sourcesCount[src] = 0;
+        }
+    }
+
+    slideSource() {
+        var globMax = 0;
+        for (var src of this._sourceNames) {
+            // res[0] = max, res[1] = slidedArray
+            var res = slide(this._sourcesArray[src], this._sourcesCount[src]);
+            // max
+            this._sourcesCountMax[src] = res[0];
+            globMax = globMax > res[0] ? globMax : res[0];
+            // data
+            this._sourcesArray[src] = res[1];
+        }
+        this._globalMax = globMax;
+    }
+
+    toArray() {
+        var to_return = [];
+        for (var src of this._sourceNames) {
+            to_return.push({
+                label: src,
+                data: this._sourcesArray[src]
+            });
+        }
+        return to_return;
+    }
+
+    getGlobalMax() {
+        return this._globalMax;
+    }
+
+    getSingleSource(sourceName) {
+        return this._sourcesArray[sourceName];
+    }
+
+    getEmptyData() {
+        return [{label: 'no data', data: emptyArray}];
+    }
+}
+
+var sources = new Sources();
+sources.addSource('global');
+
+var curNumLog = 0;
+var curMaxDataNumLog = 0;
+
 var optionsGraph = {
     series: {
                 shadowSize: 0 ,
-                lines: { fill: true, fillColor: { colors: [ { opacity: 1 }, { opacity: 0.1 } ] }}
+                lines: { 
+                    fill: true, 
+                    fillColor: { 
+                        colors: [ { opacity: 1 }, { opacity: 0.1 } ]
+                    }
+                }
             },
+    //colors: ["#2fa1db"],
     yaxis: { min: 0, max: 20 },
-    xaxis: { ticks: [[0, 0], [1, 1], [2, 2], [3, 3], [4, 4], [5, 5], [6, 6], [7, 7], [8, 8], [9, 9], [10, 10]] },
+    xaxis: { min: 0, max: numPoint-1 },
+    ticks: numPoint,
     grid: {
         tickColor: "#dddddd",
         borderWidth: 0 
@@ -39,8 +124,7 @@ $(document).ready(function () {
 
             source.onmessage = function(event) {
                 var json = jQuery.parseJSON( event.data );
-                updateLogTable(json.log);
+                updateLogTable(json.feedName, json.log);
-                //updateChartData(json.chart);
             };
         
         } else {
@@ -50,37 +134,39 @@ $(document).ready(function () {
 });
 
 
-//var plot = $.plot("#placeholder", [ [] ], optionsGraph);
+//var plot1 = $.plot("#feedDiv1", [ { label: "Number of log messages", data: dataNumLog } ], optionsGraph);
-//updateChart()
+var plot1 = $.plot("#feedDiv1", sources.getEmptyData(), optionsGraph);
+updateChart()
 
 function updateChart() {
-    plot.setData(data);
+    updateChart1();
-    plot.getOptions().yaxes[0].max = curMax[dataset];
+    updateChart2();
-    plot.setupGrid();
+    setTimeout(updateChart, updateInterval);
-    plot.draw();
-    setTimeout(update, updateInterval);
 }
 
-function updateChartData(feed) {
+function updateChart1() {
-    if (feed.length == 0)
+    sources.slideSource();
-        return;
+    sources.resetCountOnSource();
+    plot1.setData(sources.toArray());
+    plot1.getOptions().yaxes[0].max = sources.getGlobalMax();
+    plot1.setupGrid();
+    plot1.draw();
+}
 
-    for (feedName in feed) {
+function updateChart2() {
-        console.log(feedName.name);
-        if (data[feedName.name] === undefined) {
-            data[feedName.name] = {};
-        }
-        data[feedName.name].data = slide(data[feedName.name].data, feedName.data)
-    }
 
 }
 
-function updateLogTable(log) {
+function updateLogTable(feedName, log) {
     if (log.length == 0)
         return;
 
     // Create new row
     tableBody = document.getElementById('table_log_body');
+    //curNumLog++;
+    sources.addIfNotPresent(feedName);
+    sources.incCountOnSource(feedName);
+    sources.incCountOnSource('global');
     createRow(tableBody, log);
 
     // Remove old row
@@ -94,10 +180,16 @@ function updateLogTable(log) {
 }
 
 function slide(orig, newData) {
-    var slided = orig;
+    var slided = [];
-    slided.slice(newData.length);
+    var max = newData;
-    slided.concat(newData);
+    for (i=1; i<orig.length; i++) {
-    return slided
+        y = orig[i][1];
+        slided.push([i-1, y]);
+        max = y > max ? y : max;
+    }
+    slided.push([orig.length-1, newData]);
+    curMaxDataNumLog = max;
+    return [curMaxDataNumLog, slided];
 }
 
 function createRow(tableBody, log) {

templates/index.html

@@ -36,14 +36,14 @@
         <div id="page-wrapper">
             <div class="container-fluid">
                 <div class="row">
-                    <div class="col-lg-12">
+                    <div class="col-lg-6">
                         
                         <div class="panel panel-default" style="margin-top: 15px;">
                             <div class="panel-heading">
-                                <i class="fa fa-bar-chart-o fa-fw"></i> Feed
+                                <i class="fa fa-bar-chart-o fa-fw"></i> Log feed
                             </div>
                             <div id="panelbody" class="panel-body">
-                                <div id="feedDiv" style="height: 100%; position: relative;"></div>
+                                <div id="feedDiv1" style="width:100%; height:300px; position: relative;"></div>
                             </div>
                             <!-- /.panel-body -->
 
@@ -51,7 +51,24 @@
                         <!-- /.panel -->
 
                     </div>
-                    <!-- /.col-lg-12 -->
+                    <!-- /.col-lg-6 -->
+                    <!-- /.col-lg-6 -->
+                    <div class="col-lg-6">
+                        
+                        <div class="panel panel-default" style="margin-top: 15px;">
+                            <div class="panel-heading">
+                                <i class="fa fa-bar-chart-o fa-fw"></i> Feed
+                            </div>
+                            <div id="panelbody" class="panel-body">
+                                <div id="feedDiv2" style="width:100%; height:300px; position: relative;"></div>
+                            </div>
+                            <!-- /.panel-body -->
+
+                        </div>
+                        <!-- /.panel -->
+
+                    </div>
+                    <!-- /.col-lg-6 -->
                 </div>
                 <!-- /.row CHART -->
 

zmq_subscriber.py

@@ -4,25 +4,46 @@ import time
 import zmq
 import redis
 import random
+import configparser
+import os
+import sys
+import json
 
+configfile = os.path.join(os.environ['VIRTUAL_ENV'], '../config.cfg')
+cfg = configparser.ConfigParser()
+cfg.read(configfile)
+
+zmq_url = cfg.get('Redis', 'zmq_url')
 zmq_url = "tcp://crf.circl.lu:5556"
-channel = "102"
+channel = cfg.get('Redis', 'channel')
 context = zmq.Context()
 socket = context.socket(zmq.SUB)
 socket.connect(zmq_url)
 socket.setsockopt_string(zmq.SUBSCRIBE, channel)
 
 redis_server = redis.StrictRedis(
-        host='localhost',
+        host=cfg.get('Redis', 'host'),
-        port=6250,
+        port=cfg.getint('Redis', 'port'),
-        db=0)
+        db=cfg.getint('Redis', 'db'))
 
 # server side
 pubsub = redis_server.pubsub(ignore_subscribe_messages=True)
 
 while True:
+    rdm = random.randint(1,3)
+    time.sleep(float(rdm / 10))
+    content = "rdm "+str(rdm)
+    to_send = { 'name': 'feeder'+str(rdm), 'log': content }
+    redis_server.publish(channel, json.dumps(to_send))
+
+sys.exit(1)
+
+
+while True:
+    #FIXME check if sock.recv is blocking
     time.sleep(0.1)
     content = socket.recv()
+    console.log('sending')
     print(content)
     redis_server.publish(channel, content)