GDP

#!/usr/bin/env python

import sys

import gdp

import traceback

import threading

import json

import argparse

from twisted.python import log

from twisted.internet import reactor, ssl

from autobahn.twisted.websocket import WebSocketServerProtocol, \

        WebSocketServerFactory

class GDPSubscriptionProtocol(WebSocketServerProtocol):

    def __init__(self, *args):

        WebSocketServerProtocol.__init__(self, *args)

        self.conn_active = False

        self.remote = None

    def onConnect(self, request):

        print "[%s] client connected." % request.peer

        self.remote = request.peer

    def onOpen(self):

        print "[%s] connection open." % self.remote

        self.conn_active = True

    def onMessage(self, payload, isBinary):

        print "[%s] received message: %s" % (self.remote, payload)

        reactor.callInThread(self.__onMessage, payload, isBinary)

    def onClose(self, wasClean, code, reason):

        print "[%s] connection closed." % (self.remote)

        self.conn_active = False

    def __onMessage(self, payload, isBinary):

        """

        Runs in a non-reactor thread.

        payload is a dictionary to specify subscription parameters:

           {"logname": <>,      # required

            "startrec": <>,     # optional

            "numrec": <>,       # optional

           }

        """

        ## timeout of 100 ms

        ev_timeout = {"tv_sec": 0, "tv_nsec": 100 * 10**6, "tv_accuracy": 0.0}

        try:

            thread_ = threading.current_thread().getName()

            payload_dict = json.loads(payload)

            logname = payload_dict["logname"]

            startrec = payload_dict.get("startrec", 0)

            numrec = payload_dict.get("numrec", 1)

            print "[%s][%s] Subscribing to log=%s, startrec=%d, numrec=%d" % \

                              (thread_, self.remote, logname, startrec, numrec)

            lh = gdp.GDP_GIN(gdp.GDP_NAME(logname), gdp.GDP_MODE_RO)

            lh.subscribe_by_recno(startrec, numrec, None)

            while self.conn_active:

                event = None

                while event is None and self.conn_active:

                    event = lh.get_next_event(ev_timeout)

                if not self.conn_active:

                    break

                # Create a dictionary that we send to the client

                __evdict = {}

                gin_handle = event["gin"]

                event_ep_stat = event["stat"]

                assert lh == gin_handle

                datum = event["datum"]

                __evdict["logname"] = logname

                __evdict["ep_stat"] = (event_ep_stat.code,

                                        gdp.MISC.ep_stat_tostr(event_ep_stat))

                __evdict["type"] = event["type"]

                __evdict["datum"] = {"recno": datum["recno"],

                                     "data": datum["buf"].peek(),

                                     "ts": datum["ts"]}

                # send the event to client, but remove binary signature first

                ## this will have problems with binary data in log, but there

                ## is no better alternative if we want to stick to JSON

                __evstr = json.dumps(__evdict)

                print "[%s] sending event: %s" % (self.remote, __evstr)

                reactor.callFromThread(self.sendMessage, __evstr, False)

                if event["type"] == gdp.GDP_EVENT_DONE:

                    break

            ## How did we reach here.

            reason = "subcription completed" if self.conn_active \

                                                     else "client left"

            print "[%s] finished, because %s" % (self.remote, reason)

            if not self.conn_active:

                print "[%s] terminating subscription" % self.remote

                lh.unsubscribe()

        except Exception as e:

            # send any errors back to the client

            error_string = "Exception: %s\n\n" % str(e)

            error_string += traceback.format_exc()

            print "[%s] %s" % (self.remote, error_string)

            reactor.callFromThread(self.sendMessage, error_string, False)

if __name__ == "__main__":

    parser = argparse.ArgumentParser()

    parser.add_argument("-p", "--port", type=int, default=9007,

                            help="TCP port to serve requests on, default=9007")

    parser.add_argument("-s", "--secure", action='store_true',

                            help="Enable wss (listens on ws-port + 1)")

    parser.add_argument("--key", type=str,

                            help="Private key file for secure websockets")

    parser.add_argument("--cert", type=str,

                            help="Certificate file for secure websockets")

    # Get the actual arguments

    args = parser.parse_args()

    # gdp.gdp_init()

    # gdp.dbg_set("*=20")

    log.startLogging(sys.stderr)

    # Set up argument parsing code

    factory = WebSocketServerFactory()

    factory.protocol = GDPSubscriptionProtocol

    reactor.listenTCP(args.port, factory)

    if args.secure:

        contextFactory = ssl.DefaultOpenSSLContextFactory(args.key, args.cert)

        reactor.listenSSL(args.port+1, factory, contextFactory)

    reactor.run()