source: py-tcpmultiplexer/TCPMultiplexer.py@ 313

#!/usr/bin/env python
"""
# Multiplexer for HTTP streams, used for broadcast of MPJEG and WAV streams
#
# Licence: BSDLike
#
# Rick van der Zwet <info@rickvanderzwet.nl>
"""

import SocketServer
import argparse
import logging
import sys
import threading
import time
import urllib2

# Some boring defaults
DEFAULT_HOST = '0.0.0.0'
DEFAULT_PORT = 9999
DEFAULT_CONFIG = 'streams.yaml'

# URL : TARGET
DEFAULT_STREAMS = {
  '/cam1/video' : 'http://172.16.0.67:8080/videofeed',
  '/cam1/audio' : 'http://172.16.0.67:8080/audio.wav',
  }

# Global variables used as ring-buffers or shared-storage
urlheader = dict()
dataheader = dict()
urldata = dict()
running = True
recv_threads = []

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')

class ThreadedTCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
    pass


class MyTCPHandler(SocketServer.BaseRequestHandler):
    """
    The RequestHandler class for our server.

    It is instantiated once per connection to the server, and must
    override the handle() method to implement communication to the
    client.
    """

    def handle(self):
        global running, dataheader, urlheader, urldata
        # self.request is the TCP socket connected to the client
        self.data = self.request.recv(1024).strip()
        print self.data
        urlid = self.data.split('\n')[0].split()[1]
        req_type = self.data.split('\n')[0].split()[0]
        logging.info("Connection %s from '%s' for '%s'", req_type, self.client_address[0], urlid)
        try:
          if not urldata.has_key(urlid):
            self.request.send("HTTP/1.1 404 NOT FOUND\n\nThe page '%s' does not exists" % urlid)
          else:
            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)
          del urldata[urlid][self]

def get_data(url, urlid, *args):
  global running, dataheader, urlheader, urldata
  logger = logging.getLogger('recv_id=' + urlid)
  # Fill buffers if needed
  logger.info("Starting thread '%s' (%s)", url, urlid)
  while running:
    try:
      f = urllib2.urlopen(url)
      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 IOError, e:
      #Enforce a connection reset
      logger.warning("URL reset '%s' (%s)", url, e)
      del urlheader[urlid]
      del dataheader[urlid]
      del urldata[urlid]
      time.sleep(1)
      continue
  logger.info("Closing Thread '%s'", url)


if __name__ == "__main__":
  parser = argparse.ArgumentParser(description=__doc__,formatter_class=argparse.RawDescriptionHelpFormatter)
  parser.add_argument('--host', dest='host', default=DEFAULT_HOST, help='Listen to IP [default: %s]' % DEFAULT_HOST)
  parser.add_argument('--port', dest='port', default=DEFAULT_PORT, type=int, help='Listen to PORT [default: %s]' % DEFAULT_PORT)
  parser.add_argument('--stream-cfg', dest='stream_cfg', default=DEFAULT_CONFIG, help='YAML Stream configuration [default: %s]' % DEFAULT_CONFIG)
  args = parser.parse_args()

  # Inport streams
  streams = DEFAULT_STREAMS
  try:
    import yaml
    streams = yaml.load(open(args.stream_cfg))
  except (ImportError, IOError) as e:
    logging.warning("Stream config file '%s' not readable or parsable (%s)", args.stream_cfg, e)


  # Create the base server
  try:
    while True:
      try:
        ThreadedTCPServer.allow_reuse_address = True
        server = ThreadedTCPServer((args.host, args.port), MyTCPHandler)
        break
      except IOError, e:
        logging.warning('For conection %s:%s to become available (%s)', args.host, args.port , e)
        time.sleep(1)
  except KeyboardInterrupt:
    sys.exit(1)

  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', args.host, args.port)
    server.serve_forever()
  except KeyboardInterrupt, IOError:
    logging.info('Shutting down, please wait...')
    running = False
    server.shutdown()
    [thread.join() for thread in recv_threads]
    logging.info('All done, good bye!')