source: py-tcpmultiplexer/TCPMultiplexer.py@ 311

#!/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 threading
import urllib2
import time
import SocketServer
import logging
import sys

urlheaders = 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, urlheaders, 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:
          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))
          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, urlheaders, urldata
  # Fill buffers if needed
  logging.info("Starting thread '%s' (%s)" % (url, urlid))
  while running:
    try:
      f = urllib2.urlopen(url)
      urlheaders[urlid] = ''.join(f.info().headers)
      urldata[urlid] = dict()
      while running:
        data = f.read(10000)
        for key in urldata[urlid].keys():
          urldata[urlid][key].append(data)
    except UrlError:
      #Enforce a connection reset
      logging.warning("URL reset '$s'" % url)
      del urlheaders[key]
      del urldata[key]
      time.sleep(1)
      continue
  logging.info("Closing Thread '%s'" % url)


if __name__ == "__main__":
  HOST, PORT = "0.0.0.0", 9999
  # Create the server
  try:
    while True:
      try:
        ThreadedTCPServer.allow_reuse_address = True
        server = ThreadedTCPServer((HOST, PORT), MyTCPHandler)
        break
      except IOError, 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()

  # Activate the server; this will keep running until you
  # interrupt the program with Ctrl-C
  try:
    logging.info('Serving at %s:%s' % (HOST,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!')