daq: ngin-lua fetcher reconnection handling works now

1 files changed, 60 insertions, 38 deletions

diff --git a/src/daq/nginx-lua/s5-nginx-lua-fetch.py b/src/daq/nginx-lua/s5-nginx-lua-fetch.py
index 87c7bd1..4401e53 100755
--- a/src/daq/nginx-lua/s5-nginx-lua-fetch.py
+++ b/src/daq/nginx-lua/s5-nginx-lua-fetch.py
@@ -43,17 +43,18 @@ from time import sleep
 import re
 import simplejson as json
 import datetime
+import dateutil.parser
 
-_MAX_PACKET_SIZE = 8192 # in bytes
+_MAX_PACKET_SIZE = 65536 # in bytes
 
 __version__ = "$Rev$"
 
 
 class SFiveNGXluaProto(Protocol):
-    def __init__(self, finished, importer):
+    def __init__(self, finished, fetcher):
         self.data = ''
         self.finished = finished
-        self._importer = importer
+        self._fetcher = fetcher
         print 'SFive: started receiving log data'
 
     def dataReceived(self, bytes):
@@ -68,11 +69,11 @@ class SFiveNGXluaProto(Protocol):
 
             if len(lines) > 0:
                 for line in lines:
-                    self._importer.updateData(json.loads(line))
+                    self._fetcher.updateData(json.loads(line))
 
     def connectionLost(self, reason):
         print 'SFive: finished receiving log data:', reason.getErrorMessage()
-        self._importer._socketError()
+        self._fetcher.close()
 
 
 class SFivePort(unix.ConnectedDatagramPort):
@@ -102,25 +103,25 @@ class SFivePort(unix.ConnectedDatagramPort):
 
 class SFiveProto(protocol.ConnectedDatagramProtocol):
 
-    def __init__(self, importer):
-        self._importer = importer
+    def __init__(self, fetcher):
+        self._fetcher = fetcher
 
     def startProtocol(self):
-        self._importer._socketReady()
+        self._fetcher.socketReady()
 
     def datagramReceived(self, data):
         print 'SFive: received datagram: "%s" (will get ignored)' % (data)
 
     def connectionFailed(self, failure):
         print 'SFive: connection failed: %s' % (failure.getErrorMessage())
-        self._importer._socketError()
+        self._fetcher.socketError()
 
     def sendDatagram(self, data):
         try:
             return self.transport.write(data)
         except socket.error as err:
             print 'SFive: sending datagram failed: %s' % (err)
-            self._importer._socketError()
+            self._fetcher.socketError()
 
 
 class ClientList:
@@ -170,23 +171,23 @@ class NGXLuaFetcher():
 
         self._proto = None
         self._conn = None
+        self._connected = False
         self._url_re = None
-        self._reactor = reactor
 
     def run(self):
         if self._initFetch():
-            self._reactor.callWhenRunning(self._initSocket)
-            self._reactor.run()
+            reactor.callWhenRunning(self._initSocket)
+            reactor.run()
 
     def _initFetch(self):
         try:
             self._url_re = re.compile(self._nameformat)
 
             print "SFive: trying to fetch from '%s'" % self._url
-            agent = Agent(self._reactor)
+            agent = Agent(reactor)
             d = agent.request('GET', self._url, Headers({'User-Agent': ['SFive nginx-lua fetcher']}), None)
             d.addCallback(self._httpResponse)
-            d.addBoth(self._socketError)
+            d.addBoth(self.close)
 
         except re.error as e:
             print 'SFive: regex error: %s' % (e)
@@ -200,28 +201,6 @@ class NGXLuaFetcher():
         response.deliverBody(SFiveNGXluaProto(finished, self))
         return finished
 
-    def _initSocket(self):
-        print 'SFive: trying to connect to %s...' % (self._socket)
-        self._proto = SFiveProto(self)
-        self._conn = SFivePort(self._socket, self._proto, _MAX_PACKET_SIZE, 0o666, None, self._reactor)
-        self._conn.startListening()
-
-    def _socketError(self):
-        if self._conn:
-            self._conn.stopListening()
-        self._reactor.stop()
-
-    def _socketReady(self):
-        print 'SFive: connection to sfive hub established'
-        self._sendInit()
-        # TODO: start callback every self._duration seconds
-
-    def _sendInit(self):
-        initdata = { "version": 1, "hostname": self._hostname,
-                     "streamer-id": { "content-id": self._content_id, "format": self._format, "quality": self._quality },
-                     "tags": self._tags }
-        self._proto.sendDatagram('%s\n' % (json.dumps(initdata)));
-
     def updateData(self, data):
         try:
             m = re.match(self._url_re, data['url'])
@@ -234,9 +213,52 @@ class NGXLuaFetcher():
         try:
             print "url='%s' -> contend-id: '%s', format: '%s', quality: '%s'" % (data['url'],
                                             streamerid['content'], streamerid['format'], streamerid['quality'])
-            #self._clients.update(data)
+            # self._clients.update(data)
+            if self._conn and self._connected:
+                print "    --> %(time)s: from %(client)s/%(port)d (%(ua)s) %(status)d:  %(bytes-sent)d bytes" % data
+                self._sendDataset(dateutil.parser.parse(data['time']), 1, ClientList())
+
         except KeyError as e:
             print 'SFive: %s' % (e)
+        except ValueError as e:
+            print 'SFive: %s' % (e)
+        except AttributeError as e:
+            print 'SFive: %s' % (e)
+
+    def close(self):
+        if self._conn and self._connected:
+            self._conn.stopListening()
+        self._connected = False
+        if reactor.running:
+            reactor.stop()
+
+
+
+    def _initSocket(self):
+        print 'SFive: trying to connect to %s...' % (self._socket)
+        self._connected = False
+        self._proto = SFiveProto(self)
+        self._conn = SFivePort(self._socket, self._proto, _MAX_PACKET_SIZE, 0o666, None, reactor)
+        self._conn.startListening()
+
+    def socketReady(self):
+        print 'SFive: connection to sfive hub established'
+        self._connected = True
+        self._sendInit()
+        # TODO: start callback every self._duration seconds
+
+    def socketError(self):
+        print 'SFive: connection to sfive hub lost - trying reconnect'
+        if self._conn and self._connected:
+            self._conn.stopListening()
+        self._connected = False
+        reactor.callLater(3, self._initSocket)
+
+    def _sendInit(self):
+        initdata = { "version": 1, "hostname": self._hostname,
+                     "streamer-id": { "content-id": self._content_id, "format": self._format, "quality": self._quality },
+                     "tags": self._tags }
+        self._proto.sendDatagram('%s\n' % (json.dumps(initdata)));
 
     def _sendDataset(self, timestamp, duration, clients):
         data = { "start-time": timestamp.isoformat('T') + 'Z',