Implement new event marshalling system

2012-11-21 21:41:11 -05:00
parent 8ef29f3358
commit 4ac2d61359
14 changed files with 511 additions and 189 deletions
--- a/api/event-types.yml
+++ b/api/event-types.yml
@@ -0,0 +1,63 @@
+---
+types:
+  vault-slot:
+    material: int
+    quantity: int
+    damage: int
+    data: int
+    position: int
+    name: !optional string
+    durability: !optional int
+events:
+  quit:
+    player: string
+  join:
+    player: string
+  broadcast:
+    message: string
+  chat:
+    message: string
+    sender: string
+  vault-contents:
+    player: string
+    items:
+    - vault-slot
+  player-death:
+    player: string
+    message: string
+  server-heartbeat:
+    port: int
+    name: string
+    worldTimes: dict
+  web-heartbeat:
+    server: int
+    heartbeat: dict
+    time: int
+    day-period: string
+  player-kick:
+    player: string
+    message: string
+  player-message:
+    player: string
+    message: string
+  market-order:
+    orderID: int
+  block-break:
+    player: string
+    x: int
+    y: int
+    z: int
+    material: int
+  block-place:
+    player: string
+    x: int
+    y: int
+    z: int
+    material: int
+  player-murder:
+    player: string
+    killer: string
+    message: string
+  weather:
+    world: string
+    isRaining: boolean
--- a/api/events.py
+++ b/api/events.py
@@ -1,102 +1,238 @@
 from django.conf import settings
+import os
 from django.core.cache import cache
 import time
 from minecraft.models import Server
-from json import loads, dumps, JSONEncoder
+from json import loads, dumps, JSONEncoder, JSONDecoder
 import beanstalkc
 from django.contrib.auth.models import User
+import yaml
+
+class Dispatcher(object):
+  @classmethod
+  def broadcastMessage(cls, message, *args):
+    event = BroadcastEvent(message%args)
+    ServerQueue.broadcast(event)
+    WebQueue.broadcast(event)

 class EventEncoder(JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Event):
-            return {'type': obj.type, 'payload': obj.data}
+          return {'type': obj.type, 'properties': obj.properties}
+        if isinstance(obj, beanstalkc.Job):
+          return {'id': obj.id, 'event': loads(obj.body)}
        return super(EventEncoder, self).default(obj)
-            

 class Event(object):
-    def __init__(self, type, data):
+    def __init__(self, type, *args, **kwargs):
+      types = self.types()
+      if type in types:
        self.type = type
-        self.data = data
+        self.properties = self.marshallType(kwargs, types[type], type)
+      else:
+        raise ValueError, "Unknown event type '%s'"%(type)
+
+    @classmethod
+    def marshallType(cls, data, typeDescription, paramPath=""):
+      if data is None:
+        return data
+      if isinstance(typeDescription, OptionalEventProperty):
+        return cls.marshallType(data, str(typeDescription), paramPath)
+      try:
+        if typeDescription == "string":
+          return str(data)
+        if typeDescription == "int":
+          return int(data)
+        if typeDescription == "dict":
+          return dict(data)
+        if typeDescription == "list":
+          return list(data)
+      except TypeError, e:
+        raise ValueError, "Could not convert '%s' to %s type for %s"%(data,
+            typeDescription, paramPath)
+      if isinstance(typeDescription, dict):
+        ret = {}
+        for key, elementType in typeDescription.iteritems():
+          key = key.replace('-', '_')
+          if key in data:
+            ret[key] = cls.marshallType(data[key], elementType,
+                "%s.%s"%(paramPath, key))
+          elif not isinstance(elementType, OptionalEventProperty):
+            raise KeyError, "Missing parameter '%s.%s'"%(paramPath, key)
+        return ret
+      if isinstance(typeDescription, list):
+        ret = []
+        i = 0
+        for item in data:
+          ret.append(cls.marshallType(item, typeDescription[0], "%s[%d]"%(paramPath, i)))
+          i += 1
+        return ret
+      if typeDescription in cls.datatypes():
+        return cls.marshallType(data, cls.datatypes()[typeDescription], paramPath)
+      raise TypeError, "Unknown event data type '%s.%s'"%(paramPath, typeDescription)
+
+    @classmethod
+    def types(cls):
+      types = cache.get('caminus-event-types')
+      if types is None:
+        typefile = os.path.sep.join(
+          __file__.split(os.path.sep)[0:-1]+
+          ['event-types.yml',]
+        )
+        yaml.add_constructor(r'!optional', OptionalEventProperty)
+        types = yaml.load(open(typefile, 'r'))['events']
+        cache.set('caminus-event-types', types, 3600)
+      return types
+
+    @classmethod
+    def datatypes(cls):
+      typefile = os.path.sep.join(
+        __file__.split(os.path.sep)[0:-1]+
+        ['event-types.yml',]
+      )
+      yaml.add_constructor(r'!optional', OptionalEventProperty)
+      types = yaml.load(open(typefile, 'r'))['types']
+      return types
+
    def __repr__(self):
-      return json.dumps(self, cls=EventEncoder)
+      return dumps(self, cls=EventEncoder)

-class ChatEvent(Event):
-    def __init__(self, sender, message):
-      super(ChatEvent, self).__init__(type='chat', data={'sender': sender,
-        'message': message})
+    def toJSON(self):
+      return dumps(self, cls=EventEncoder)

-class VaultContentsEvent(Event):
-    def __init__(self, player, items):
-      super(VaultContentsEvent, self).__init__(type='vault-contents',
-          data={'player': player, 'items': items})
+    @classmethod
+    def fromJSON(cls, json):
+      decoded = loads(json, cls=EventDecoder)
+      print decoded
+      evt = Event(json['type'], json['event']['payload'])

-class QuitEvent(Event):
-  def __init__(self, player):
-    super(QuitEvent, self).__init__(type='quit', data={'player': player})
+class OptionalEventProperty(str):
+  def __new__(cls, loader, node):
+    return str.__new__(cls, loader.construct_scalar(node))

-class JoinEvent(Event):
-  def __init__(self, player):
-    super(JoinEvent, self).__init__(type='join', data={'player': player})
-
-class BroadcastEvent(Event):
-    def __init__(self, message):
-        super(BroadcastEvent, self).__init__(type='broadcast', data={'message':
-          message})
-
-class PlayerDeathEvent(Event):
-    def __init__(self, player, message):
-        super(PlayerDeathEvent, self).__init__(type='player-death',
-            data={'player': player, 'message': message})
-
-class ServerHeartbeatEvent(Event):
-    def __init__(self, server, data):
-        now = server.current_time()
-        t = now.second+now.minute*60+now.hour*60*60
-        super(ServerHeartbeatEvent, self).__init__(type='server-heartbeat',
-            data={'server': server.id, 'heartbeat': data, 'time': t,
-              'day-period': server.day_period()})
-
-class KickEvent(Event):
-    def __init__(self, user, message):
-        super(KickEvent, self).__init__(type='player-kick',
-            data={'player': user, 'message': message})
-
-class PlayerMessageEvent(Event):
-    def __init__(self, user, message):
-        super(PlayerMessageEvent, self).__init__(type='player-message',
-            data={'message': message, 'player': user})
-
-class MarketOrderEvent(Event):
-    def __init__(self, orderID):
-        super(MarketOrderEvent, self).__init__(type='market-order',
-            data={'orderID': orderID})
-
-def server_queue(server, users=[]):
-    queueName = 'caminus-broadcast-%s'%server.id
-    queue = beanstalkc.Connection(host=settings.CAMINUS_BEANSTALKD_HOST,
+class EventQueue(object):
+  def __init__(self, name):
+    self.name = name
+    self.queue = beanstalkc.Connection(host=settings.CAMINUS_BEANSTALKD_HOST,
        port=settings.CAMINUS_BEANSTALKD_PORT)
-    queue.use(queueName)
-    queue.watch(queueName)
-    if len(users) > 0:
-        for user in users:
-            queue.watch("caminus-user-%s"%user)
-    return queue
+    self.queue.use(self.name)
+    self.queue.watch(self.name)

-def send_server_event(server, event):
-    if settings.CAMINUS_USE_BEANSTALKD:
-        queue = server_queue(server)
-        json = dumps(event, cls=EventEncoder)
-        queue.put(json)
+  def sendEvent(self, evt):
+    assert(isinstance(evt, Event))
+    eventData = {}
+    eventData['stamp'] = time.time()
+    eventData['event'] = evt
+    json = dumps(eventData, cls=EventEncoder)
+    self.queue.put(json)

-def broadcast_server_event(event):
+  def getEvents(self, timeout=30):
+    ret = []
+    job = self.queue.reserve(timeout)
+    while job:
+      ret.append(job)
+      job = self.queue.reserve(timeout=0)
+    return ret
+
+  def flush(self):
+    job = self.queue.peek_ready()
+    while job:
+      job.delete()
+      job = self.queue.peek_ready()
+
+  @staticmethod
+  def connection():
+    return beanstalkc.Connection(host=settings.CAMINUS_BEANSTALKD_HOST,
+        port=settings.CAMINUS_BEANSTALKD_PORT)
+
+  @classmethod
+  def allQueues(cls):
+    c = cls.connection()
+    ret = []
+    for c in c.tubes():
+      ret.append(EventQueue(c))
+    return ret
+
+  def deleteJob(self, id):
+    self.queue.delete(id)
+
+class MarketQueue(EventQueue):
+  def __init__(self):
+    super(MarketQueue, self).__init__('caminus-market')
+
+class ServerQueue(EventQueue):
+  def __init__(self, server):
+    super(ServerQueue, self).__init__('caminus-broadcast-%s'%server.id)
+
+  @classmethod
+  def broadcast(cls, evt):
    for server in Server.objects.all():
-        send_server_event(server, event)
+      queue = ServerQueue(server)
+      queue.sendEvent(evt)
+
+class WebQueue(EventQueue):
+  def __init__(self, id):
+    self.id = id
+    super(WebQueue, self).__init__('caminus-web-%s'%(id))
+    if id != "0":
+      activeCache = cache.get('minecraft-web-queues')
+      if activeCache is None:
+        activeCache = {}
+      activeCache[id] = time.time()
+      expired = []
+      for queueName, stamp in activeCache.iteritems():
+        if time.time()-stamp > 120:
+          expired.append(queueName)
+          del activeCache[queueName]
+      cache.set('minecraft-web-queues', activeCache, 3600)
+      for e in expired:
+        q = WebQueue(e)
+        q.flush()
+
+  @classmethod
+  def broadcast(cls, event):
+    latest = cache.get('minecraft-web-events')
+    if latest is None:
+      latest = []
+    latest.append(dumps(event, cls=EventEncoder))
+    while len(latest) > 10:
+      latest.pop(0)
+    cache.set('minecraft-web-events', latest, 86400)
+
+    for queue in cls.activeQueues():
+      queue.sendEvent(event)
+
+  @staticmethod
+  def activeQueues():
+    activeCache = cache.get('minecraft-web-queues')
+    if activeCache is None:
+      activeCache = {}
+    ret = []
+    for name in activeCache.iterkeys():
+      ret.append(WebQueue(name))
+    return ret
+
+  def getEvents(self):
+    if self.id == "0":
+      latestEvents = cache.get('minecraft-web-events')
+      if latestEvents is None:
+        latestEvents = []
+      ret = []
+      for e in latestEvents:
+        ret.append(loads(e))
+      return ret
+    else:
+      ret = super(WebQueue, self).getEvents()
+      for evt in ret:
+        evt.delete()
+      return ret

 def server_broadcast(message, *args):
    message = message%args
    for server in Server.objects.all():
-      event = BroadcastEvent(message)
-      send_server_event(server, event)
+      event = Event('broadcast', message=message)
+      queue = ServerQueue(server)
+      queue.sendEvent(event)
    send_web_event(event)

 def user_message(user, message, *args):
@@ -107,36 +243,8 @@ def player_message(playername, message, *args):
    message = message%args
    for server in Server.objects.all():
      event = PlayerMessageEvent(playername, message)
-      send_server_event(server, event)
-
-def web_queue(id):
-    queueName = 'caminus-web-%s'%id
-    queue = beanstalkc.Connection(host=settings.CAMINUS_BEANSTALKD_HOST,
-        port = settings.CAMINUS_BEANSTALKD_PORT)
-    queue.use(queueName)
-    queue.watch(queueName)
-    activeCache = cache.get('minecraft-web-queues')
-    if activeCache is None:
-      activeCache = {}
-    activeCache[id] = time.time()
-    for queueName,stamp in activeCache.iteritems():
-      if time.time()-stamp > 120:
-        del activeCache[queueName]
-    cache.set('minecraft-web-queues', activeCache, 3600)
-    return queue
-
-def market_queue():
-    queueName = 'caminus-market'
-    queue = beanstalkc.Connection(host=settings.CAMINUS_BEANSTALKD_HOST,
-        port = settings.CAMINUS_BEANSTALKD_PORT)
-    queue.use(queueName)
-    queue.watch(queueName)
-    return queue
-
-def queue_market_event(event):
-  queue = market_queue()
-  json = dumps({'stamp': time.time(), 'event': event}, cls=EventEncoder)
-  queue.put(json)
+      queue = ServerQueue(server)
+      queue.sendEvent(event)

 def send_web_event(event):
   latest = cache.get('minecraft-web-events')
@@ -151,7 +259,7 @@ def send_web_event(event):
         port = settings.CAMINUS_BEANSTALKD_PORT)
     json = dumps({'stamp': time.time(), 'event':event}, cls=EventEncoder)
     activeQueues = cache.get('minecraft-web-queues')
-     print "Active queues", repr(activeQueues)
+     #print "Active queues", repr(activeQueues)
     if activeQueues is None:
       activeQueues = {}
     for tube in activeQueues.iterkeys():
@@ -169,4 +277,4 @@ def send_web_event(event):

 def chat(playername, message):
  evt = ChatEvent(playername, message)
-  send_web_event(evt)
+  WebQueue.broadcast(evt)
--- a/api/handlers.py
+++ b/api/handlers.py
@@ -12,7 +12,7 @@ from urllib2 import urlopen
 import json
 from datetime import datetime
 from models import cachePlayerList
-from events import server_queue, web_queue, chat, server_broadcast, send_web_event, QuitEvent, JoinEvent, PlayerDeathEvent, ServerHeartbeatEvent
+from events import ServerQueue, WebQueue, Event, chat
 from bounty.models import Bounty
 from vault.models import VaultSlot

@@ -44,7 +44,7 @@ class NewPlayerSessionHandler(BaseHandler):
            server = request.server
            profile = MinecraftProfile.objects.get(mc_username__exact=playername)
            session = PlayerSession.objects.create(server=server, player=profile, ip=ip)
-            send_web_event(JoinEvent(playername))
+            WebQueue.broadcast(Event('join', player=playername))
            return {'success': True, 'error': '', 'permissions': profile.serverPermissions(), 'sessionId': session.id}
        else:
            return {'success': False, 'error': 'Your account is inactive.', 'permissions': []}
@@ -57,7 +57,7 @@ class ClosePlayerSessionHandler(BaseHandler):
        for session in sessions:
            session.end = datetime.now()
            session.save()
-            send_web_event(QuitEvent(playername))
+            WebQueue.broadcast(Event('quit', player=playername))
        return {'valid': True}

 class EconomyHandler(BaseHandler):
@@ -88,23 +88,21 @@ class ServerEventHandler(BaseHandler):
    allowed_methods = ('GET', 'POST', 'PUT')

    def read(self, request):
-        queue = server_queue(request.server)
-        queue.watch('caminus-broadcast-%s'%request.server.id)
-        events = []
-        job = queue.reserve(timeout=30)
-        while job:
-          job.bury()
-          events.append({'id': job.jid, 'event': json.loads(job.body)})
-          job = queue.reserve(timeout=0)
-        return {'events': events, 'is-live': settings.CAMINUS_USE_BEANSTALKD}
+      queue = ServerQueue(request.server)
+      events = []
+      for e in queue.getEvents():
+        events.append(json.loads(e.body)['event'])
+        e.bury()
+      print {'events': events}
+      return {'events': events, 'is-live': settings.CAMINUS_USE_BEANSTALKD}

    def create(self, request):
-        queue = server_queue(request.server)
-        try:
-            queue.delete(int(request.POST['job']))
-        except Exception, e:
-            pass
-        return {'result': 'success'}
+      queue = ServerQueue(request.server)
+      try:
+        queue.deleteJob(int(request.POST['job']))
+      except Exception, e:
+        pass
+      return {'result': 'success'}

    def update(self, request):
        events = json.loads(request.POST['events'])['events']
@@ -112,18 +110,21 @@ class ServerEventHandler(BaseHandler):
            if evt['type'] == 'chat':
                chat(evt['payload']['sender'], evt['payload']['message'])
            if evt['type'] == 'player-death':
-              send_web_event(PlayerDeathEvent(evt['payload']['player'],
-              evt['payload']['message']))
-            if evt['type'] == 'heartbeat':
+              WebQueue.broadcast(Event('player-death', player=evt['payload']['player'],
+              message=evt['payload']['message']))
+            if evt['type'] == 'server-heartbeat':
              cache.set('caminus-server-heartbeat-%s'%(request.server.id), evt['payload'], 86400)
-              send_web_event(ServerHeartbeatEvent(request.server, evt['payload']))
+              WebQueue.broadcast(Event('web-heartbeat', server=request.server.id,
+                heartbeat=evt['payload'],
+                day_period=request.server.day_period(),
+                time=request.server.current_time().second))
            if evt['type'] == 'player-murder':
              bounties = Bounty.objects.filter(target__mc_username=evt['payload']['player'])
              killer = MinecraftProfile.objects.get(mc_username=evt['payload']['killer'])
              for bounty in bounties:
                bounty.close(killer)
              if len(bounties) > 0:
-                server_broadcast("The bounty on %s has been collected."%(evt['payload']['player']))
+                Dispatcher.broadcastMessage("The bounty on %s has been collected."%(evt['payload']['player']))
        return {'result': 'success'}

 class ChatHandler(BaseHandler):
@@ -131,8 +132,8 @@ class ChatHandler(BaseHandler):

    def create(self, request):
      chat(request.user.minecraftprofile.mc_username, request.POST['message'])
-      server_broadcast("<%s> %s"%(request.user.minecraftprofile.mc_username,
-        request.POST['message']))
+      ServerQueue.broadcast(BroadcastEvent("<%s> %s",
+        request.user.minecraftprofile.mc_username, request.POST['message']))

 class PollHandler(BaseHandler):
    allowed_methods = ('GET',)
@@ -150,20 +151,8 @@ class PollHandler(BaseHandler):
        pollData['events'] = []
        pollData['info'] = info
        pollData['poll-id'] = timestamp
-        if timestamp == "0" and settings.CAMINUS_USE_BEANSTALKD:
-          pollData['poll-id'] = time.time()
-          latestEvents = cache.get('minecraft-web-events')
-          if not latestEvents:
-            latestEvents = []
-          for e in latestEvents:
-            pollData['events'].append(json.loads(e))
-        elif settings.CAMINUS_USE_BEANSTALKD:
-          eventQueue = web_queue(timestamp)
-          event = eventQueue.reserve(timeout=30)
-          while event:
-            pollData['events'].append(json.loads(event.body))
-            event.delete()
-            event = eventQueue.reserve(timeout=0)
+        eventQueue = WebQueue(timestamp)
+        pollData['events'] = eventQueue.getEvents()
        return pollData

 class VaultHandler(BaseHandler):
--- a/api/management/commands/event_stats.py
+++ b/api/management/commands/event_stats.py
@@ -9,16 +9,16 @@ class Command(BaseCommand):
  def handle(self, *args, **options):
    servers = Server.objects.all()
    for s in servers:
-      queue = events.server_queue(s)
-      stats = queue.stats()
+      queue = events.ServerQueue(s)
+      stats = queue.queue.stats()
      print s
      for k,v in stats.iteritems():
        print "\t%s: %s"%(k, v)
    print "Tubes:"
-    for t in queue.tubes():
+    for t in queue.queue.tubes():
      print "\t%s"%(t)
-      queue.use(t)
-      next = queue.peek_ready()
+      queue.queue.use(t)
+      next = queue.queue.peek_ready()
      if next:
        print "\t\tNext job: %s"%(next.body)
      else:
--- a/api/management/commands/event_types.py
+++ b/api/management/commands/event_types.py
@@ -0,0 +1,140 @@
+from django.core.management.base import BaseCommand
+from django.core.cache import cache
+from api import events
+from pprint import pprint
+from optparse import make_option
+
+_t = type
+
+class Command(BaseCommand):
+  help = 'Display information about available event types'
+  option_list = BaseCommand.option_list + (
+    make_option('--header',
+      help='Header to append to each file'),
+    make_option('--output',
+      help='Directory to write files to',
+      default='.'),
+    make_option('--lang',
+      help='Language to export to'),
+  )
+
+  def handle(self, *args, **options):
+    if options['lang'] == 'java':
+      if options['output']:
+        outdir = options['output']
+      else:
+        outdir = '.'
+      header = ""
+      if options['header']:
+        fh = open(options['header'], 'r')
+        header = fh.read()
+      for typename, properties in events.Event.types().iteritems():
+        outfile = open(outdir+'/'+self.javaName(typename)+"Event.java", 'w')
+        outfile.write(header)
+        outfile.write("import org.json.*;\n")
+        outfile.write("import java.util.*;\n")
+        outfile.write("public class %sEvent extends Event {\n"%(self.javaName(typename)))
+        for propname,type in properties.iteritems():
+          outfile.write("\tpublic %s %s;\n"%(self.javaType(type), self.javaName(propname, False)))
+        outfile.write("\tpublic static %sEvent fromJSON(JSONObject obj) throws JSONException {\n"%(self.javaName(typename)))
+        outfile.write("\t\t%sEvent ret = new %sEvent();\n"%(self.javaName(typename), self.javaName(typename)))
+        for propname, type in properties.iteritems():
+          if isinstance(type, events.OptionalEventProperty):
+            outfile.write("\t\ttry {\n\t")
+          outfile.write("\t\t"+self.jsonCode(propname, type)+"\n");
+          if isinstance(type, events.OptionalEventProperty):
+            outfile.write("\t\t} catch (JSONException e) {}\n")
+        outfile.write("\t\treturn ret;\n")
+        outfile.write("\t}\n")
+        outfile.write("\tpublic JSONWriter toJSON(JSONWriter writer) throws JSONException {\n")
+        for propname, type in properties.iteritems():
+          outfile.write("writer.key(\"%s\").value(this.%s);\n"%(propname, self.javaName(propname, False)))
+        outfile.write("\t\treturn writer;\n")
+        outfile.write("\t}\n")
+        outfile.write("\tpublic String jsonName() { return \"%s\"; }\n"%(typename))
+        outfile.write("\tstatic { Event.registerHandler(\"%s\", %sEvent.class); }\n"%(typename, self.javaName(typename)))
+        outfile.write("}\n")
+        outfile.close()
+        print "Wrote", self.javaName(typename)+"Event.java"
+      for typename, properties in events.Event.datatypes().iteritems():
+        outfile = open(outdir+'/'+self.javaName(typename)+".java", 'w')
+        outfile.write(header)
+        outfile.write("import org.json.*;\n")
+        outfile.write("import java.util.*;\n")
+        outfile.write("public class %s {\n"%(self.javaName(typename)))
+        for propname, type in properties.iteritems():
+          outfile.write("\tpublic %s %s;\n"%(self.javaType(type), self.javaName(propname, False)))
+        outfile.write("\tpublic static %s fromJSON(JSONObject obj) throws JSONException {\n"%(self.javaName(typename)))
+        outfile.write("\t\t%s ret = new %s();\n"%(self.javaName(typename), self.javaName(typename)))
+        for propname, type in properties.iteritems():
+          if isinstance(type, events.OptionalEventProperty):
+            outfile.write("\t\ttry {\n\t")
+          outfile.write("\t\t"+self.jsonCode(propname, type)+"\n");
+          if isinstance(type, events.OptionalEventProperty):
+            outfile.write("\t\t} catch (JSONException e) {}\n")
+        outfile.write("\t\treturn ret;\n")
+        outfile.write("\t}\n")
+        outfile.write("}\n")
+        outfile.close()
+        print "Wrote", self.javaName(typename)+".java"
+    else:
+      print 'Unknown language', options['lang']
+
+  @staticmethod
+  def javaName(name, isClass=True):
+    name = ''.join(map(lambda x:x[0].upper()+x[1:], name.split('-')))
+    if not isClass:
+      name = name[0].lower()+name[1:]
+    return ''.join(name)
+
+  @classmethod
+  def jsonCode(cls, propname, type):
+    propname = cls.javaName(propname, False)
+    tmpl = "ret.%s = obj.%s(\"%s\");"
+    if type == "string":
+      return tmpl%(propname, "getString", propname)
+    if type == "int":
+      return tmpl%(propname, "getInt", propname)
+    if type == "boolean":
+      return tmpl%(propname, "getBoolean", propname)
+    if type == "dict":
+      return """
+      JSONObject %s_obj = obj.getJSONObject("%s");
+      for(String name : JSONObject.getNames(%s_obj)) {
+        ret.%s.put(name, %s_obj.get(name));
+      }
+      """%(propname, propname, propname, propname, propname)
+    if isinstance(type, str):
+      return "ret.%s = %s.fromJSON(obj.getJSONObject(%s));"%(propname,
+        cls.javaName(type), propname)
+    if isinstance(type, list):
+      return """
+      JSONArray %s_list = obj.getJSONArray("%s");
+      ret.%s = new ArrayList<%s>();
+      for(int i = 0;i<%s_list.length();i++) {
+        ret.%s.add(%s.fromJSON(%s_list.getJSONObject(i)));
+      }"""%(
+        propname, propname, propname, cls.javaName(type[0]), propname, propname,
+        cls.javaName(type[0]), propname
+      )
+    raise TypeError, "Cannot convert %s to java type from JSON."%(type)
+
+  @classmethod
+  def javaType(cls, type):
+    if type == "string":
+      return "String"
+    if type == "int":
+      return "int"
+    if type == "list":
+      return "List<?>"
+    if type == "dict":
+      return "Map<String, Object>"
+    if type == "boolean":
+      return "boolean"
+    if isinstance(type, str):
+      return cls.javaName(type)
+    if isinstance(type, list):
+      return "List<%s>"%(cls.javaType(type[0]))
+    if isinstance(type, dict):
+      return "Map<String, Object>"
+    raise TypeError, "Cannot convert %s to java type."%(type)
--- a/api/management/commands/flush_events.py
+++ b/api/management/commands/flush_events.py
@@ -8,12 +8,12 @@ class Command(BaseCommand):
  def handle(self, *args, **options):
    servers = Server.objects.all()
    for s in servers:
-      queue = events.server_queue(s)
-      stats = queue.stats()
-      for t in queue.tubes():
-        queue.use(t)
-        job = queue.peek_ready()
+      queue = events.ServerQueue(s)
+      stats = queue.queue.stats()
+      for t in queue.queue.tubes():
+        queue.queue.use(t)
+        job = queue.queue.peek_ready()
        while job:
          print "Deleting %s from %s: %s"%(job.jid, t, job.body)
          job.delete()
-          job = queue.peek_ready()
+          job = queue.queue.peek_ready()
--- a/api/tests.py
+++ b/api/tests.py
@@ -8,6 +8,26 @@ import hashlib
 import events
 from django.conf import settings

+class EventTypeTest(unittest.TestCase):
+  def setUp(self):
+    pass
+
+  def tearDown(self):
+    pass
+
+  def testValidTypes(self):
+    e = events.Event('quit', player="TestPlayer")
+    self.assertEqual(e.player, "TestPlayer")
+
+  def testInvalidType(self):
+    e = events.Event('invalid')
+
+  def testMissingArgs(self):
+    e = events.Event('quit')
+
+  def testExtraArgs(self):
+    e = events.Event('quit', player='TestPlayer', extra='extra')
+
 class ServerPingTest(unittest.TestCase):
    def setUp(self):
        self.client = Client()
--- a/market/management/commands/market_processor.py
+++ b/market/management/commands/market_processor.py
@@ -1,7 +1,7 @@
 from django.core.management.base import NoArgsCommand
 from market.models import MarketOrder
 import json
-from api.events import market_queue
+from api.events import MarketQueue
 from datetime import datetime

 class Command(NoArgsCommand):
@@ -11,27 +11,27 @@ class Command(NoArgsCommand):
    print "Processing currently open orders"
    for order in MarketOrder.objects.filter(close_stamp__isnull=True):
      order.process()
-    queue = market_queue()
+    queue = MarketQueue()
    while True:
-      job = queue.reserve()
-      jobInfo = json.loads(job.body)
-      if jobInfo['event']['type'] == "market-order":
-        try:
-          order = MarketOrder.objects.get(id=jobInfo['event']['payload']['orderID'])
-        except MarketOrder.DoesNotExist:
-          # The orders might not be saved yet due to transactions
-          job.release()
-          continue
-        if order.close_stamp:
-          print "Got event for a closed order", order.id
-          job.delete()
-          continue
-        if order.quantity == 0:
-          print "Saving bugged order", order.id
-          order.close_stamp = datetime.now()
-          order.save()
-          job.delete()
-          continue
-        order.process()
+      for job in queue.getEvents():
+        jobInfo = json.loads(job.body)
+        if jobInfo['event']['type'] == "market-order":
+          try:
+            order = MarketOrder.objects.get(id=jobInfo['event']['payload']['orderID'])
+          except MarketOrder.DoesNotExist:
+            # The orders might not be saved yet due to transactions
+            job.release()
+            continue
+          if order.close_stamp:
+            print "Got event for a closed order", order.id
+            job.delete()
+            continue
+          if order.quantity == 0:
+            print "Saving bugged order", order.id
+            order.close_stamp = datetime.now()
+            order.save()
+            job.delete()
+            continue
+          order.process()

-      job.delete()
+        job.delete()
--- a/market/models.py
+++ b/market/models.py
@@ -4,7 +4,7 @@ from django.db import transaction
 from django.db.models import F
 from datetime import datetime
 from django.db.models.signals import post_save
-from api.events import queue_market_event, MarketOrderEvent
+from api.events import MarketQueue, Event
 from vault.models import VaultSlot
 from django.contrib.auth.models import User
 from minecraft.items import ITEMS
@@ -68,8 +68,9 @@ class MarketOrder(models.Model):

  def enqueue_process(self):
    if settings.CAMINUS_USE_BEANSTALKD:
-      evt = MarketOrderEvent(self.id)
-      queue_market_event(evt)
+      evt = Event('market-order', orderID=self.id)
+      queue = MarketQueue()
+      queue.sendEvent(evt)
    else:
      self.process()

--- a/minecraft/models.py
+++ b/minecraft/models.py
@@ -178,5 +178,5 @@ import api.events
 def kick_banned_user(sender, instance, created, **kwargs):
    player = instance.player
    if player.isBanned():
-      api.events.broadcast_server_event(api.events.KickEvent(player.mc_username, instance.reason))
+      api.events.ServerQueue.broadcast(api.events.KickEvent(player.mc_username, instance.reason))
 post_save.connect(kick_banned_user, sender=Ban)
--- a/1
+++ b/1
@@ -11,3 +11,4 @@ pydot==1.0.28
 beanstalkc
 stripe
 django-devserver
+PyYAML==3.10
--- a/settings.py
+++ b/settings.py
@@ -96,6 +96,7 @@ TEMPLATE_LOADERS = (
 )

 MIDDLEWARE_CLASSES = (
+    'caminus.devtools.PrintExceptionMiddleware',
    'django.middleware.common.CommonMiddleware',
    'django.contrib.sessions.middleware.SessionMiddleware',
    'django.middleware.csrf.CsrfViewMiddleware',
@@ -104,7 +105,6 @@ MIDDLEWARE_CLASSES = (
    'django.contrib.flatpages.middleware.FlatpageFallbackMiddleware',
    'django.middleware.http.ConditionalGetMiddleware',
    'django.middleware.gzip.GZipMiddleware',
-    'caminus.devtools.PrintExceptionMiddleware',
 )

 ROOT_URLCONF = 'caminus.urls'
--- a/static/js/poll.js
+++ b/static/js/poll.js
@@ -78,8 +78,8 @@ EventPoller.prototype._successfulPoll = function(data) {

  var that = this;
  $(data['events']).each(function(idx, evt) {
-    console.log(evt['event']['type']);
-    that.dispatchEvent(evt['event']);
+    console.log(evt['type']);
+    that.dispatchEvent(evt);
  });

  if (data['is-live']) {
--- a/vault/models.py
+++ b/vault/models.py
@@ -1,7 +1,7 @@
 from django.db import models
 from minecraft.models import MinecraftProfile
 from django.db.models.signals import post_save, post_delete
-from api.events import VaultContentsEvent, broadcast_server_event, send_web_event
+from api.events import Event, ServerQueue, send_web_event
 from minecraft.models import Item
 from django.conf import settings

@@ -63,7 +63,7 @@ def send_vault_update(sender, instance, created, *args, **kwargs):
  if instance.item:
    slots = [
      {
-        'item': instance.item.material,
+        'material': instance.item.material,
        'quantity': instance.quantity,
        'damage': instance.item.damage,
        'data': instance.item.data,
@@ -75,7 +75,7 @@ def send_vault_update(sender, instance, created, *args, **kwargs):
  else:
    slots = [
      {
-        'item': None,
+        'material': None,
        'quantity': None,
        'damage': None,
        'data': None,
@@ -84,8 +84,8 @@ def send_vault_update(sender, instance, created, *args, **kwargs):
        'durability': None
      }
    ]
-  evt = VaultContentsEvent(instance.player.mc_username, slots)
-  broadcast_server_event(evt)
+  evt = Event('vault-contents', player=instance.player.mc_username, items=slots)
+  ServerQueue.broadcast(evt)
  send_web_event(evt)

 post_save.connect(send_vault_update, sender=VaultSlot, dispatch_uid='derp')