gdamjan/eventsource.py

## eventsource.py
#! /usr/bin/env python2
'''\
This is a reimplementation of a node.js demo of the Server-Sent Events API.
You can find (and compare to) the node.js version at https://github.com/remy/eventsource-h5d/.
'''


from werkzeug import Request, Response, redirect
from werkzeug.wsgi import SharedDataMiddleware

from gevent import pywsgi, spawn, sleep
from gevent.queue import Queue

import os, time
import json

coroutines = []
history = []       # really should be a bound FIFO list


def update_loadavg():
    while True:
        sleep(1)
        with open("/proc/loadavg") as f:
            line = f.readline()
            loadavg = line.split()[:3]
        data = dict(zip(("1","5","15"), loadavg))
        broadcast('uptime', data)

def update_time():
    while True:
        sleep(1)
        broadcast('time', time.time());


lastMessageId = 0

def broadcast(event, data):
    global lastMessageId
    lastMessageId += 1
    message = {
      'id': lastMessageId,
      'event': event,
      'data': json.dumps(data)
    }
    # keep history, but only the last 100 messages
    history.append(message)
    del history[:-100]
    # now notify all active clients
    for q in coroutines:
        q.put(message)


totalRequests = 0

def handle_stats(req):
    if req.headers.get('accept') != 'text/event-stream':
        return Response('Not implemented', status=501)

    def event_source_iter():
        last_event_id = req.headers.get('last-event-id', type=int)
        if last_event_id:
            # replay history
            for message in history:
                if message['id'] >= last_event_id:
                    yield 'event: %(event)s\ndata: %(data)s\nid: %(id)s\n\n' % message
        else:
            yield 'id\n\n'

        q = Queue()
        try:
            coroutines.append(q)
            broadcast('requests', totalRequests)
            broadcast('connections', len(coroutines))
            for message in q:
                yield 'event: %(event)s\ndata: %(data)s\nid: %(id)s\n\n' % message
        finally:
            coroutines.remove(q)
            broadcast('connections', len(coroutines))

    headers = [('cache-control', 'no-cache'), ('connection', 'keep-alive')]
    return Response(event_source_iter(), headers=headers,
        content_type='text/event-stream')

# for wsgi (uwsgi etc)
application = Request.application(handle_stats)

# for standalone
@Request.application
def app(req):
    global totalRequests
    totalRequests += 1
    broadcast('requests', totalRequests)
    # fixup file serving etc...
    if req.path in ('/index.html', '/'):
        return redirect('/eventsource-h5d.html')
    if req.path == '/stats':
        return handle_stats(req)
    return Response('Not Found', status=404)


if __name__ == '__main__':
    host, port = '0.0.0.0', 8000
    spawn(update_loadavg)
    spawn(update_time)
    print "Serving on http://%s:%d/" % (host, port)
    # poor mans static file serving
    app = SharedDataMiddleware(app, {'/': os.path.dirname(__file__)})
    server = pywsgi.WSGIServer((host,port), app)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass

## index.html
<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Server-Sent Events: server monitor</title>
  <style>
    body { font-family: monospace; font-size: 18px; margin: 0; }
    #connStatus { background: #c00; color: #fff; font-weight: bold; }
    #connStatus.open { background: #0c0; }
    p { padding: 10px 20px; margin: 0; }
    canvas { margin: 10px 20px; border-bottom: 1px solid #ccc; border-top: 1px solid #eee; }
  </style>
</head>
<body>
  <p id="connStatus">Waiting to connect to server...</p>
  <p>Total connected clients: <span id="connections">0</span></p>
  <p>Total requests: <span id="requests">0</span></p>
  <p>Load: 1 min: <span id="l1">0</span>, 5 min: <span id="l5">0</span>, 15 min: <span id="l15">0</span></p>
  <ol id="debug"></ol>
  <canvas id="spark"></canvas>
  <script>
  if (!window.console) {
    console = {
      log: function (s) {
        document.getElementById('debug').innerHTML += '<li><pre>' + s + '</pre></li>';
      }
    };
  }
  </script>
  <script src="sse.js"></script>
</body>
</html>

## sse.js
(function () {
var connStatus = document.getElementById('connStatus'),
    connections = document.getElementById('connections'),
    requests = document.getElementById('requests'),
    load = { 1: document.getElementById('l1'), 5: document.getElementById('l5'), 15: document.getElementById('l15') },
    hasCanvas = !!document.createElement('canvas').getContext;

function connectionOpen(open) {
  connStatus.className = open ? 'open' : '';
  connStatus.innerHTML = open ? 'Active connection to server' : 'Connection dropped - trying to reopen';
}

function updateConnections(event) {
  connections.innerHTML = JSON.parse(event.data);
}

function updateRequests(event) {
  requests.innerHTML = JSON.parse(event.data);
}

var lastL1 = null,
    history = [];

function updateUptime(event) {
  var loadData = JSON.parse(event.data);
  for (var key in loadData) {
    load[key].innerHTML = loadData[key];
  }

  if (hasCanvas) {
    // normalise
    var l1 = (loadData[1] * 100 | 0) + 0.5;

    history.unshift(l1);
    if (history.length > 400) {
      history.splice(400, 400 - history.length);
    }

    var max = Math.max.apply(Math, history) * 1.25;

    ctx.clearRect(0, 0, ctx.canvas.width, ctx.canvas.height);
    ctx.save();
    ctx.fillStyle = '#000';
    ctx.fillText(loadData[1], 0, 50 - (history[0]/max * 50));
    ctx.restore();
    ctx.beginPath();
    for (var i = 0; i < history.length; i++) {
      ctx.lineTo(i + 20.5, 50 - (history[i-1]/max * 50) + 0.5);
    }
    ctx.stroke();
    ctx.closePath();
  }
}

var source = new EventSource('/stats');
source.addEventListener('open', function () { connectionOpen(true); }, false);
source.addEventListener('error', function () { connectionOpen(false); }, false);
source.addEventListener('connections', updateConnections, false);
source.addEventListener('requests', updateRequests, false);
source.addEventListener('uptime', updateUptime, false);
source.addEventListener('time', function (e) {  }, false);

var ctx;

if (hasCanvas) {
  ctx = document.getElementById('spark').getContext('2d');
  ctx.canvas.height = 50;
  ctx.canvas.width = 400;
  ctx.fillStyle = '#259BDA';
  ctx.strokeStyle = '#259BDA';
}

})();
	#! /usr/bin/env python2
	'''\
	This is a reimplementation of a node.js demo of the Server-Sent Events API.
	You can find (and compare to) the node.js version at https://github.com/remy/eventsource-h5d/.
	'''


	from werkzeug import Request, Response, redirect
	from werkzeug.wsgi import SharedDataMiddleware

	from gevent import pywsgi, spawn, sleep
	from gevent.queue import Queue

	import os, time
	import json

	coroutines = []
	history = [] # really should be a bound FIFO list


	def update_loadavg():
	while True:
	sleep(1)
	with open("/proc/loadavg") as f:
	line = f.readline()
	loadavg = line.split()[:3]
	data = dict(zip(("1","5","15"), loadavg))
	broadcast('uptime', data)

	def update_time():
	while True:
	sleep(1)
	broadcast('time', time.time());


	lastMessageId = 0

	def broadcast(event, data):
	global lastMessageId
	lastMessageId += 1
	message = {
	'id': lastMessageId,
	'event': event,
	'data': json.dumps(data)
	}
	# keep history, but only the last 100 messages
	history.append(message)
	del history[:-100]
	# now notify all active clients
	for q in coroutines:
	q.put(message)



	totalRequests = 0

	def handle_stats(req):
	if req.headers.get('accept') != 'text/event-stream':
	return Response('Not implemented', status=501)

	def event_source_iter():
	last_event_id = req.headers.get('last-event-id', type=int)
	if last_event_id:
	# replay history
	for message in history:
	if message['id'] >= last_event_id:
	yield 'event: %(event)s\ndata: %(data)s\nid: %(id)s\n\n' % message
	else:
	yield 'id\n\n'

	q = Queue()
	try:
	coroutines.append(q)
	broadcast('requests', totalRequests)
	broadcast('connections', len(coroutines))
	for message in q:
	yield 'event: %(event)s\ndata: %(data)s\nid: %(id)s\n\n' % message
	finally:
	coroutines.remove(q)
	broadcast('connections', len(coroutines))

	headers = [('cache-control', 'no-cache'), ('connection', 'keep-alive')]
	return Response(event_source_iter(), headers=headers,
	content_type='text/event-stream')

	# for wsgi (uwsgi etc)
	application = Request.application(handle_stats)

	# for standalone
	@Request.application
	def app(req):
	global totalRequests
	totalRequests += 1
	broadcast('requests', totalRequests)
	# fixup file serving etc...
	if req.path in ('/index.html', '/'):
	return redirect('/eventsource-h5d.html')
	if req.path == '/stats':
	return handle_stats(req)
	return Response('Not Found', status=404)


	if __name__ == '__main__':
	host, port = '0.0.0.0', 8000
	spawn(update_loadavg)
	spawn(update_time)
	print "Serving on http://%s:%d/" % (host, port)
	# poor mans static file serving
	app = SharedDataMiddleware(app, {'/': os.path.dirname(__file__)})
	server = pywsgi.WSGIServer((host,port), app)
	try:
	server.serve_forever()
	except KeyboardInterrupt:
	pass
	<!DOCTYPE html>
	<html lang="en">
	<head>
	<meta charset="utf-8">
	<title>Server-Sent Events: server monitor</title>
	<style>
	body { font-family: monospace; font-size: 18px; margin: 0; }
	#connStatus { background: #c00; color: #fff; font-weight: bold; }
	#connStatus.open { background: #0c0; }
	p { padding: 10px 20px; margin: 0; }
	canvas { margin: 10px 20px; border-bottom: 1px solid #ccc; border-top: 1px solid #eee; }
	</style>
	</head>
	<body>
	<p id="connStatus">Waiting to connect to server...</p>
	<p>Total connected clients: <span id="connections">0</span></p>
	<p>Total requests: <span id="requests">0</span></p>
	<p>Load: 1 min: <span id="l1">0</span>, 5 min: <span id="l5">0</span>, 15 min: <span id="l15">0</span></p>
	<ol id="debug"></ol>
	<canvas id="spark"></canvas>
	<script>
	if (!window.console) {
	console = {
	log: function (s) {
	document.getElementById('debug').innerHTML += '<li><pre>' + s + '</pre></li>';
	}
	};
	}
	</script>
	<script src="sse.js"></script>
	</body>
	</html>
	(function () {
	var connStatus = document.getElementById('connStatus'),
	connections = document.getElementById('connections'),
	requests = document.getElementById('requests'),
	load = { 1: document.getElementById('l1'), 5: document.getElementById('l5'), 15: document.getElementById('l15') },
	hasCanvas = !!document.createElement('canvas').getContext;

	function connectionOpen(open) {
	connStatus.className = open ? 'open' : '';
	connStatus.innerHTML = open ? 'Active connection to server' : 'Connection dropped - trying to reopen';
	}

	function updateConnections(event) {
	connections.innerHTML = JSON.parse(event.data);
	}

	function updateRequests(event) {
	requests.innerHTML = JSON.parse(event.data);
	}

	var lastL1 = null,
	history = [];

	function updateUptime(event) {
	var loadData = JSON.parse(event.data);
	for (var key in loadData) {
	load[key].innerHTML = loadData[key];
	}

	if (hasCanvas) {
	// normalise
	var l1 = (loadData[1] * 100 \| 0) + 0.5;

	history.unshift(l1);
	if (history.length > 400) {
	history.splice(400, 400 - history.length);
	}

	var max = Math.max.apply(Math, history) * 1.25;

	ctx.clearRect(0, 0, ctx.canvas.width, ctx.canvas.height);
	ctx.save();
	ctx.fillStyle = '#000';
	ctx.fillText(loadData[1], 0, 50 - (history[0]/max * 50));
	ctx.restore();
	ctx.beginPath();
	for (var i = 0; i < history.length; i++) {
	ctx.lineTo(i + 20.5, 50 - (history[i-1]/max * 50) + 0.5);
	}
	ctx.stroke();
	ctx.closePath();
	}
	}

	var source = new EventSource('/stats');
	source.addEventListener('open', function () { connectionOpen(true); }, false);
	source.addEventListener('error', function () { connectionOpen(false); }, false);
	source.addEventListener('connections', updateConnections, false);
	source.addEventListener('requests', updateRequests, false);
	source.addEventListener('uptime', updateUptime, false);
	source.addEventListener('time', function (e) { }, false);

	var ctx;

	if (hasCanvas) {
	ctx = document.getElementById('spark').getContext('2d');
	ctx.canvas.height = 50;
	ctx.canvas.width = 400;
	ctx.fillStyle = '#259BDA';
	ctx.strokeStyle = '#259BDA';
	}

	})();