code.grnet.gr Git - flowspy/blob - poller/views.py

   1 # -*- coding: utf-8 -*- vim:fileencoding=utf-8:
   2 # vim: tabstop=4:shiftwidth=4:softtabstop=4:expandtab
   3
   4 # Copyright © 2011-2014 Greek Research and Technology Network (GRNET S.A.)
   5 # Copyright © 2011-2014 Leonidas Poulopoulos (@leopoul)
   6 #
   7 # Permission to use, copy, modify, and/or distribute this software for any
   8 # purpose with or without fee is hereby granted, provided that the above
   9 # copyright notice and this permission notice appear in all copies.
  10 #
  11 # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH REGARD
  12 # TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND
  13 # FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR
  14 # CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE,
  15 # DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS
  16 # ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS
  17 # SOFTWARE.
  18
  19 from gevent.pool import Pool
  20 import gevent
  21 import json
  22
  23 import uuid
  24 import datetime
  25 from django.shortcuts import render_to_response
  26 from django.template.loader import render_to_string
  27 from django.http import HttpResponse
  28 from gevent.event import Event
  29 from django.conf import settings
  30 #from django.views.decorators.csrf import csrf_exempt
  31 from django.http import HttpResponseRedirect
  32 from django.core.urlresolvers import reverse
  33 from django.conf import settings
  34
  35 import beanstalkc
  36
  37 import logging
  38 import os
  39
  40 LOG_FILENAME = os.path.join(settings.LOG_FILE_LOCATION, 'poller.log')
  41 formatter = logging.Formatter('%(asctime)s %(levelname)s: %(message)s')
  42 logger = logging.getLogger(__name__)
  43 logger.setLevel(logging.DEBUG)
  44 handler = logging.FileHandler(LOG_FILENAME)
  45 handler.setFormatter(formatter)
  46 logger.addHandler(handler)
  47
  48 def create_message(message, user, time):
  49     data = {'id': str(uuid.uuid4()), 'body': message, 'user':user, 'time':time}
  50     data['html'] = render_to_string('poll_message.html', dictionary={'message': data})
  51     return data
  52
  53
  54 def json_response(value, **kwargs):
  55     kwargs.setdefault('content_type', 'text/javascript; charset=UTF-8')
  56     return HttpResponse(json.dumps(value), **kwargs)
  57
  58 class Msgs(object):
  59     cache_size = 500
  60
  61     _instance = None
  62     def __new__(cls, *args, **kwargs):
  63         if not cls._instance:
  64             cls._instance = super(Msgs, cls).__new__(cls, *args, **kwargs)
  65         return cls._instance
  66
  67     def __init__(self):
  68         logger.info("initializing")
  69         self.user = None
  70         self.user_cache = {}
  71         self.user_cursor = {}
  72         self.cache = []
  73         self.new_message_event = None
  74         self.new_message_user_event = {}
  75
  76     def main(self, request):
  77         if self.user_cache:
  78             request.session['cursor'] = self.user_cache[-1]['id']
  79         return render_to_response('poll.html', {'messages': self.user_cache})
  80
  81     def message_existing(self, request):
  82         if request.is_ajax():
  83             try:
  84                 user = request.user.get_profile().peer.peer_tag
  85             except:
  86                 user = None
  87                 return False
  88             try:
  89                 assert(self.new_message_user_event[user])
  90             except:
  91                 self.new_message_user_event[user] = Event()
  92             try:
  93                 if self.user_cache[user]:
  94                     self.user_cursor[user] = self.user_cache[user][-1]['id']
  95             except:
  96                 self.user_cache[user] = []
  97                 self.user_cursor[user] = ''
  98             return json_response({'messages': self.user_cache[user]})
  99         return HttpResponseRedirect(reverse('group-routes'))
 100
 101     def message_new(self, mesg=None):
 102         if mesg:
 103             message = mesg['message']
 104             user = mesg['username']
 105             logger.info("from %s" %user)
 106             now = datetime.datetime.now()
 107             msg = create_message(message, user, now.strftime("%Y-%m-%d %H:%M:%S"))
 108         try:
 109             isinstance(self.user_cache[user], list)
 110         except:
 111             self.user_cache[user] = []
 112         self.user_cache[user].append(msg)
 113         if self.user_cache[user][-1] == self.user_cache[user][0]:
 114             self.user_cursor[user] = self.user_cache[user][-1]['id']
 115         else:
 116             self.user_cursor[user] = self.user_cache[user][-2]['id']
 117         if len(self.user_cache[user]) > self.cache_size:
 118             self.user_cache[user] = self.user_cache[user][-self.cache_size:]
 119         try:
 120             assert(self.new_message_user_event[user])
 121         except:
 122             self.new_message_user_event[user] = Event()
 123         self.new_message_user_event[user].set()
 124         self.new_message_user_event[user].clear()
 125         return json_response(msg)
 126
 127     def message_updates(self, request):
 128         if request.is_ajax():
 129             cursor = {}
 130             try:
 131                 user = request.user.get_profile().peer.peer_tag
 132             except:
 133                 user = None
 134                 return False
 135             try:
 136                 cursor[user] = self.user_cursor[user]
 137             except:
 138                 return HttpResponse(content='', mimetype=None, status=400)
 139
 140             try:
 141                 if not isinstance(self.user_cache[user], list):
 142                     self.user_cache[user] = []
 143             except:
 144                 self.user_cache[user] = []
 145             if not self.user_cache[user] or cursor[user] == self.user_cache[user][-1]['id']:
 146                 self.new_message_user_event[user].wait(settings.POLL_SESSION_UPDATE)
 147             try:
 148                 for index, m in enumerate(self.user_cache[user]):
 149                     if m['id'] == cursor[user]:
 150                         return json_response({'messages': self.user_cache[user][index + 1:]})
 151                 return json_response({'messages': self.user_cache[user]})
 152             finally:
 153                 if self.user_cache[user]:
 154                     self.user_cursor[user] = self.user_cache[user][-1]['id']
 155         return HttpResponseRedirect(reverse('group-routes'))
 156
 157     def monitor_polls(self):
 158         b = beanstalkc.Connection()
 159         b.watch(settings.POLLS_TUBE)
 160         while True:
 161             job = b.reserve()
 162             msg = json.loads(job.body)
 163             job.bury()
 164             logger.info("Got New message")
 165             self.message_new(msg)
 166
 167
 168     def start_polling(self):
 169         logger.info("Start Polling")
 170         gevent.spawn(self.monitor_polls)
 171
 172
 173 msgs = Msgs()
 174 main = msgs.main
 175
 176 message_updates = msgs.message_updates
 177 message_existing = msgs.message_existing
 178
 179
 180 poll = msgs.start_polling
 181 poll()