webstreamr.tac

from twisted.application import internet, service
from twisted.web import server
from twisted.internet.defer import DeferredQueue
from twisted.web.resource import Resource
from twisted.internet import protocol
import simplejson

PORT = 80

listeners = {}

class Queue(DeferredQueue):
    def __init__(self, handler=None):
        DeferredQueue.__init__(self)
        self.message_handler = handler
        def f(msg):
            if self.message_handler:
                self.message_handler(msg)
            else:
                self.receivedMessage(msg)
            self.get().addCallback(f)
        self.get().addCallback(f)

class StreamResource(Resource):
    isLeaf = True
    
    def render_POST(self, request):
        if not request.path in listeners:
            listeners[request.path] = []
        
        data = dict(request.args)
        for k in data:
            if len(data[k]) < 2:
                data[k] = data[k][0]
        
        for listener in listeners[request.path]:
            listener.queue.put(simplejson.dumps(data))
            
        return "OK :: Sent to %s listeners\n" % len(listeners[request.path])
    
    def render_GET(self, request):
        if not request.path in listeners:
            listeners[request.path] = []
           
        request.jsonp = request.args.get('callback', [None])[0] 
        request.queue = Queue(lambda m: self._send(request, m))
        listeners[request.path].append(request)
            
        if request.jsonp:
            request.setHeader('Content-Type', 'text/javascript')
        else:
            request.setHeader('Content-Type', 'application/json')
        request.setHeader('Transfer-Encoding', 'chunked')
        request.notifyFinish().addBoth(self._finished, request)
        
        return server.NOT_DONE_YET
    
    def _finished(self, whatever, request=None):
        listeners[request.path].remove(request)

    def _send(self, request, message):
        if request.jsonp:
            request.write("%s('%s');\n" % (request.jsonp, message))
            request.finish()
        else:
            request.write("%s\n" % message)
   

class PolicyProtocol(protocol.Protocol):
    def dataReceived(self, line):
        if "/>" in line:
            self.transport.write("<cross-domain-policy>\n<site-control permitted-cross-domain-policies=\"all\"/>\n<allow-access-from domain=\"*\" to-ports=\"%s\"/>\n</cross-domain-policy>\n" % PORT)
            self.transport.loseConnection()
    
class PolicyFactory(protocol.ServerFactory):
    protocol = PolicyProtocol

application = service.Application('streamr')
serviceCollection = service.IServiceCollection(application)
internet.TCPServer(PORT, server.Site(StreamResource())).setServiceParent(serviceCollection)
internet.TCPServer(843, PolicyFactory()).setServiceParent(serviceCollection)