Changeset 312

Timestamp:

Jul 19, 2011, 12:38:06 PM (13 years ago)

Author:

Rick van der Zwet

Message:

Making sure we have valid starting headers...

File:

• py-tcpmultiplexer/TCPMultiplexer.py (modified) (4 diffs)

py-tcpmultiplexer/TCPMultiplexer.py

@@ -7 +7 @@
 # Rick van der Zwet <info@rickvanderzwet.nl>
 #
-import threading
-import urllib2
-import time
 import SocketServer
+import argparse
 import logging
 import sys
+import threading
+import time
+import urllib2
 
-urlheaders = dict()
+# Some boring defaults
+DEFAULT_HOST = '0.0.0.0'
+DEFAULT_PORT = '9999'
+
+# Global variables used as ring-buffers or shared-storage
+urlheader = dict()
+dataheader = dict()
 urldata = dict()
 running = True
@@ -35 +42 @@
 
     def handle(self):
-        global running, urlheaders, urldata
+        global running, dataheader, urlheader, urldata
         # self.request is the TCP socket connected to the client
         self.data = self.request.recv(1024).strip()
@@ -43 +50 @@
         logging.info("Connection %s from '%s' for '%s'", req_type, self.client_address[0], urlid)
         try:
-          self.request.send('HTTP/1.1 200 OK\n' + urlheaders[urlid] + '\n')
-          if req_type == 'HEAD':
-            return
-          elif req_type == 'GET':
-            urldata[urlid][self] = []
-            while running:
-              if len(urldata[urlid][self]) == 0:
-                time.sleep(0.1)
-                continue
-              self.request.send(urldata[urlid][self].pop(0))
+          if not urldata.has_key(urlid):
+            self.request.send("HTTP/1.1 404 NOT FOUND\n\nThe page '%s' does not exists" % urlid)
           else:
-             assert False, "Invalid request"
+            self.request.send('HTTP/1.1 200 OK\n' + urlheader[urlid] + 'X-Proxy-Server: TCPMultiplexer.py\n' + '\n' + dataheader[urlid])
+            if req_type == 'HEAD':
+              return
+            elif req_type == 'GET':
+              urldata[urlid][self] = []
+              while running:
+                if len(urldata[urlid][self]) == 0:
+                  time.sleep(0.1)
+                  continue
+                self.request.send(urldata[urlid][self].pop(0))
+            else:
+               assert False, "Invalid request"
         except IOError:
           pass
         except StandardError, e:
-          logging.warning('Server has issues: %s' % e)
+          logging.warning('Server has issues: %s', e)
           del urldata[urlid][self]
 
 def get_data(url, urlid, *args):
-  global running, urlheaders, urldata
+  global running, dataheader, urlheader, urldata
+  logger = logging.getLogger('recv_id=' + urlid)
   # Fill buffers if needed
-  logging.info("Starting thread '%s' (%s)" % (url, urlid))
+  logger.info("Starting thread '%s' (%s)", url, urlid)
   while running:
     try:
       f = urllib2.urlopen(url)
-      urlheaders[urlid] = ''.join(f.info().headers)
+      urlheader[urlid] = ''.join(f.info().headers)
       urldata[urlid] = dict()
+      datatype = None
+      urltype = f.info().gettype()
+      logger.info("url.info().gettype() is '%s'", urltype)
+      if urltype == 'audio/x-wav':
+        datatype = 'wav'
+      elif urltype == 'multipart/x-mixed-replace':
+        datatype = 'mjeg'
+      else:
+        logger.warning("Cannot find file type of '%s'", url)
+
+      if datatype:
+        logger.info("'%s' Identified as %s", url, datatype)
+
+      if datatype == 'wav':
+        dataheader[urlid] = f.read(44)
+      elif datatype == 'mjeg':
+        dataheader[urlid] = '\n'.join(f.read(100).split('\n')[0:2])
+      else:
+        dataheader[urlid] = ''
+
       while running:
         data = f.read(10000)
         for key in urldata[urlid].keys():
           urldata[urlid][key].append(data)
-    except UrlError:
+    except IOError, e:
       #Enforce a connection reset
-      logging.warning("URL reset '$s'" % url)
-      del urlheaders[key]
-      del urldata[key]
+      logger.warning("URL reset '%s' (%s)", url, e)
+      del urlheader[urlid]
+      del dataheader[urlid]
+      del urldata[urlid]
       time.sleep(1)
       continue
-  logging.info("Closing Thread '%s'" % url)
+  logger.info("Closing Thread '%s'", url)
 
 
 if __name__ == "__main__":
+  parser = argparse.ArgumentParser(description='Process some integers.')
+
+  # URL : TARGET
+  DEFAULT_STREAMS = {
+    '/cam1/video' : 'http://172.16.0.67:8080/videofeed',
+    '/cam1/audio' : 'http://172.16.0.67:8080/audio.wav',
+    }
+  # Flexible enough to specify using yaml file (sys.argv) as well
+  if len(sys.argv) > 1:
+    config_file = sys.argv[1]
+    try:
+      import yaml
+      streams = yaml.load(open(config_file))
+    except (ImportError, IOError) as e:
+      logging.error("Config file '%s' not readable or parsable (%s)", config_file, e)
+
+
   HOST, PORT = "0.0.0.0", 9999
   # Create the server
@@ -94 +143 @@
         break
       except IOError, e:
-        logging.warning('For conection %s:%s to become available (%s)' % (HOST, PORT, e))
+        logging.warning('For conection %s:%s to become available (%s)', HOST, PORT, e)
         time.sleep(1)
   except KeyboardInterrupt:
     sys.exit(1)
 
-  url = 'http://172.16.0.67:8080/videofeed'
-  recv_threads.append(threading.Thread(target=get_data, args=(url,'/video1')))
-  recv_threads[-1].setDaemon(True)
-  recv_threads[-1].start()
+  for urlid, url in streams.iteritems():
+    recv_threads.append(threading.Thread(target=get_data, args=(url,urlid)))
+    recv_threads[-1].setDaemon(True)
+    recv_threads[-1].start()
 
   # Activate the server; this will keep running until you
   # interrupt the program with Ctrl-C
   try:
-    logging.info('Serving at %s:%s' % (HOST,PORT))
+    logging.info('Serving at %s:%s', HOST,PORT)
     server.serve_forever()
   except KeyboardInterrupt, IOError: