qwebirc/ajaxengine.py

   1 from twisted.web import resource, server, static
   2 from twisted.names import client
   3 from twisted.internet import reactor
   4 import traceback
   5 import simplejson, md5, sys, os, ircclient, time, config, weakref
   6
   7 Sessions = {}
   8
   9 def get_session_id():
  10   return md5.md5(os.urandom(16)).hexdigest()[:10]
  11
  12 class BufferOverflowException(Exception):
  13   pass
  14
  15 def jsondump(fn):
  16   def decorator(*args, **kwargs):
  17     x = fn(*args, **kwargs)
  18     if isinstance(x, list):
  19       return simplejson.dumps(x)
  20     return x
  21   return decorator
  22
  23 def cleanupSession(id):
  24   try:
  25     del Sessions[id]
  26   except KeyError:
  27     pass
  28
  29 class IRCSession:
  30   def __init__(self, id):
  31     self.id = id
  32     self.subscriptions = []
  33     self.buffer = []
  34     self.throttle = 0
  35     self.schedule = None
  36     self.closed = False
  37     self.cleanupschedule = None
  38
  39   def subscribe(self, channel):
  40     if len(self.subscriptions) >= config.MAXSUBSCRIPTIONS:
  41       self.subscriptions.pop(0)
  42
  43     self.subscriptions.append(channel)
  44     self.flush()
  45
  46   def flush(self, scheduled=False):
  47     if scheduled:
  48       self.schedule = None
  49
  50     if not self.buffer or not self.subscriptions:
  51       return
  52
  53     t = time.time()
  54
  55     if t < self.throttle:
  56       if not self.schedule:
  57         self.schedule = reactor.callLater(self.throttle - t, self.flush, True)
  58       return
  59     else:
  60       # process the rest of the packet
  61       if not scheduled:
  62         if not self.schedule:
  63           self.schedule = reactor.callLater(0, self.flush, True)
  64         return
  65
  66     self.throttle = t + config.UPDATE_FREQ
  67
  68     encdata = simplejson.dumps(self.buffer)
  69     self.buffer = []
  70
  71     newsubs = []
  72     for x in self.subscriptions:
  73       if x.write(encdata):
  74         newsubs.append(x)
  75
  76     self.subscriptions = newsubs
  77     if self.closed and not self.subscriptions:
  78       cleanupSession(self.id)
  79
  80   def event(self, data):
  81     bufferlen = sum(map(len, self.buffer))
  82     if bufferlen + len(data) > config.MAXBUFLEN:
  83       self.buffer = []
  84       self.client.error("Buffer overflow")
  85       return
  86
  87     self.buffer.append(data)
  88     self.flush()
  89
  90   def push(self, data):
  91     if not self.closed:
  92       self.client.write(data)
  93
  94   def disconnect(self):
  95     # keep the session hanging around for a few seconds so the
  96     # client has a chance to see what the issue was
  97     self.closed = True
  98
  99     reactor.callLater(5, cleanupSession, self.id)
 100
 101 class Channel:
 102   def __init__(self, request):
 103     self.request = request
 104
 105 class SingleUseChannel(Channel):
 106   def write(self, data):
 107     self.request.write(data)
 108     self.request.finish()
 109     return False
 110
 111 class MultipleUseChannel(Channel):
 112   def write(self, data):
 113     self.request.write(data)
 114     return True
 115
 116 class AJAXEngine(resource.Resource):
 117   isLeaf = True
 118
 119   def __init__(self, prefix):
 120     self.prefix = prefix
 121
 122   @jsondump
 123   def render_POST(self, request):
 124     path = request.path[len(self.prefix):]
 125     if path == "/n":
 126       ip = request.transport.getPeer()
 127       ip = ip[1]
 128
 129       nick, ident = request.args.get("nick"), "webchat"
 130       if not nick:
 131         return [False, "Nickname not supplied"]
 132
 133       nick = nick[0]
 134
 135       id = get_session_id()
 136
 137       session = IRCSession(id)
 138
 139       client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=nick)
 140       session.client = client
 141
 142       Sessions[id] = session
 143
 144       return [True, id]
 145     return [False, "404"]
 146
 147   @jsondump
 148   def render_GET(self, request):
 149     path = request.path[len(self.prefix):]
 150     if path.startswith("/s/"):
 151       sessionid = path[3:]
 152       session = Sessions.get(sessionid)
 153
 154       if not session:
 155         return [False, "Bad session ID"]
 156
 157       session.subscribe(SingleUseChannel(request))
 158       return server.NOT_DONE_YET
 159     if path.startswith("/p/"):
 160       command = request.args.get("c")
 161       if not command:
 162         return [False, "No command specified"]
 163
 164       command = command[0]
 165
 166       sessionid = path[3:]
 167       session = Sessions.get(sessionid)
 168       if not session:
 169         return [False, "Bad session ID"]
 170
 171       try:
 172         decoded = command.decode("utf-8")
 173       except UnicodeDecodeError:
 174         decoded = command.decode("iso-8859-1", "ignore")
 175
 176       try:
 177         session.push(decoded)
 178       except AttributeError: # occurs when we haven't noticed an error
 179         session.disconnect()
 180         return [False, "Connection closed by server."]
 181       except Exception, e: # catch all
 182         session.disconnect()
 183         traceback.print_exc(file=sys.stderr)
 184         return [False, "Unknown error."]
 185
 186       return [True]
 187
 188     return [False, "404"]