python http streaming via multisocketsink with gstreamer

[Alexander Heinz]

Hello,

I would like to stream a single source to multiple clients via http. 
Clients should be able to connect and disconnect while streaming is 
already ongoing.

I am aware of http launch but I would prefer to do this in python 
because I would like to add additional functionality later that easier 
for me to write in python.

I use the gstreamer packages that come with raspbian Jessie (GStreamer 
1.4.4) on a Raspberry Pi 2

My questions:
- I cannot stream to more than one client with the code below and I 
would like to know what I need to change to make it work
- If I stop my single client and try to reconnect afterwards, the stream 
does not start again. What needs to be done to fix this?
- the code does not even work for a single client if I use raspbian 
stretch (which comes with gstreamer 1.10.4). Any idea what needs to be 
changed?

Thank you for your help!

Best regards,
Alex

P.S.:
here is the code:

#!/usr/bin/env python
import os,signal,sys,time
import gi,BaseHTTPServer,SocketServer

gi.require_version('Gst', '1.0')
from gi.repository import GObject, Gio, Gst

# Initialize the gstreamer toolchain
Gst.init(None)
GObject.threads_init()

# reopen stdout in non-buffered mode
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)


class gst_broadcaster:

     def __init__(self, gpipeline):
         self.pipeline = Gst.parse_launch(gpipeline)
         self.source = self.pipeline.get_by_name('source')
         self.sink = self.pipeline.get_by_name('sink')
         self.bus = self.pipeline.get_bus()
         self.bus.add_signal_watch()
         self.bus.connect('message',self.message_handler)
         self.sink.connect('client-added',self.add_client)
         self.sink.connect('client-socket-removed',self.remove_client)
         self.sink.set_property('timeout',30000000000)
         self.clients = []
         self.pipeline.set_state(Gst.State.READY)

     def kill(self):
         self.pipeline.set_state(Gst.State.NULL)

     def add_client(self,sink,gsock):
         self.pipeline.set_state(Gst.State.PLAYING)
         self.clients.append(gsock)

     def remove_client(self,sink,gsock):
         self.sink.emit('remove-flush',gsock)
         if gsock in self.clients:
             self.clients.remove(gsock)

     def handle(self, wfile):
         gsock = Gio.Socket().new_from_fd(wfile.fileno())
         self.sink.emit("add", gsock)
         while gsock in self.clients:
             time.sleep(1)

     def message_handler(self,bus,message):
         msgType = message.type
         if msgType == Gst.MessageType.ERROR:
             self.kill()
             print "\n Unable to play Video. Error: ", \
             message.parse_error()
         elif msgType == Gst.MessageType.EOS:
             self.kill()

def get_pipeline():
     r= "videotestsrc name=source is-live=true ! x264enc "
     r+= ' ! h264parse ! matroskamux name=mux streamable=true ! 
multisocketsink name=sink'
     return r

def get_player():
     pipeline = get_pipeline()
     player =  gst_broadcaster(pipeline)
     return player

class MyHTTPHandler(BaseHTTPServer.BaseHTTPRequestHandler):

     def do_HEAD(self):
         self.send_response(200)
         self.send_header("Content-type", "video/x-matroska")
         self.end_headers()
     def do_GET(self):
         self.do_HEAD()
         player.handle(self.wfile)


class ThreadedHTTPServer(SocketServer.ThreadingMixIn, 
BaseHTTPServer.HTTPServer):
     """Handle requests in a separate thread."""

port = 8001
httpd = ThreadedHTTPServer(("0.0.0.0", port), MyHTTPHandler)

def httpd_start():
     print time.asctime(), "server start (port %i)" % port
     httpd.serve_forever()

def httpd_shutdown():
     httpd.server_close()
     print time.asctime(), "server stop"

def signal_handler(signal, frame):
     try:
         httpd_shutdown()
     finally:
       sys.exit(0)


if __name__ == '__main__':
     signal.signal(signal.SIGINT, signal_handler)
     signal.signal(signal.SIGTERM, signal_handler)
     player = get_player()
     try:
         httpd_start()
     except KeyboardInterrupt:
         pass
     httpd_shutdown()