Source Code for Module flumotion.component.consumers.hlsstreamer.hlsstreamer

  1  # -*- Mode: Python -*- 
  2  # vi:si:et:sw=4:sts=4:ts=4 
  3   
  4  # Flumotion - a streaming media server 
  5  # Copyright (C) 2004,2005,2006,2007,2008,2009 Fluendo, S.L. 
  6  # Copyright (C) 2010,2011 Flumotion Services, S.A. 
  7  # All rights reserved. 
  8  # 
  9  # This file may be distributed and/or modified under the terms of 
 10  # the GNU Lesser General Public License version 2.1 as published by 
 11  # the Free Software Foundation. 
 12  # This file is distributed without any warranty; without even the implied 
 13  # warranty of merchantability or fitness for a particular purpose. 
 14  # See "LICENSE.LGPL" in the source distribution for more information. 
 15  # 
 16  # Headers in this file shall remain intact. 
 17   
 18  import gst 
 19  import urlparse 
 20   
 21  from twisted.internet import reactor 
 22   
 23  from flumotion.common import gstreamer 
 24  from flumotion.common.i18n import gettexter 
 25  from flumotion.component.base import http 
 26  from flumotion.component.component import moods 
 27  from flumotion.component.common.streamer.fragmentedstreamer import\ 
 28          FragmentedStreamer, Stats 
 29  from flumotion.component.consumers.hlsstreamer.resources import \ 
 30          HTTPLiveStreamingResource 
 31  from flumotion.component.consumers.hlsstreamer.hlsring import HLSRing 
 32  from flumotion.component.consumers.hlsstreamer import hlssink 
 33   
 34  __all__ = ['HLSStreamer'] 
 35  __version__ = "" 
 36  T_ = gettexter() 
 37   
 38   
 39  SUPPORTED_FORMATS = {"video/mpegts": ("video/mpegts", "video/mpegts", "ts"), 
 40                       "video/webm": ("video/webm", "video/webm", "webm")} 
 41   
 42   
 43 -class HLSStreamer(FragmentedStreamer, Stats): 
 44      DEFAULT_SESSION_TIMEOUT = 30 
 45      DEFAULT_FRAGMENT_PREFIX = 'fragment' 
 46      DEFAULT_MAIN_PLAYLIST = 'main.m3u8' 
 47      DEFAULT_STREAM_PLAYLIST = 'stream.m3u8' 
 48      DEFAULT_STREAM_BITRATE = 300000 
 49      DEFAULT_KEYFRAMES_PER_SEGMENT = 10 
 50   
 51      logCategory = 'hls-streamer' 
 52   
 53      _mime_type = None 
 54      _content_type = None 
 55      _stream_setup = False 
 56   
 57 -    def init(self): 
 58          self.debug("HTTP live streamer initialising") 
 59          self.hlsring = None 
 60   
 61 -    def get_mime(self): 
 62          return self._mime_type 
 63   
 64 -    def get_content_type(self): 
 65          return self._content_type 
 66   
 67 -    def get_pipeline_string(self, properties): 
 68          # Always use the python element for now. The C element will be used 
 69          # when it is mature enough. 
 70          # if not gstreamer.element_factory_exists('hlssink'): 
 71          hlssink.register() 
 72          return "hlssink name=sink sync=false" 
 73   
 74 -    def configure_auth_and_resource(self): 
 75          self.httpauth = http.HTTPAuthentication(self) 
 76          self.resource = HTTPLiveStreamingResource(self, self.httpauth, 
 77                  self.secret_key, self.session_timeout) 
 78   
 79 -    def getRing(self): 
 80          return self.hlsring 
 81   
 82 -    def configure_pipeline(self, pipeline, props): 
 83          self.hlsring = HLSRing( 
 84              props.get('main-playlist', self.DEFAULT_MAIN_PLAYLIST), 
 85              props.get('stream-playlist', self.DEFAULT_STREAM_PLAYLIST), 
 86              props.get('stream-bitrate', self.DEFAULT_STREAM_BITRATE), 
 87              self.description, 
 88              props.get('fragment-prefix', self.DEFAULT_FRAGMENT_PREFIX), 
 89              props.get('new-fragment-tolerance', 0), 
 90              props.get('max-window', self.DEFAULT_MAX_WINDOW), 
 91              props.get('max-extra-buffers', None), 
 92              props.get('key-rotation', 0), 
 93              props.get('keys-uri', None)) 
 94   
 95          # Call the base class after initializing the ring and getting 
 96          # the secret key and the session timeout 
 97          FragmentedStreamer.configure_pipeline(self, pipeline, props) 
 98   
 99          self.hls_url = props.get('hls-url', None) 
100          if self.hls_url: 
101              if not self.hls_url.endswith('/'): 
102                  self.hls_url += '/' 
103              if self.mountPoint.startswith('/'): 
104                  mp = self.mountPoint[1:] 
105              else: 
106                  mp = self.mountPoint 
107              self.hls_url = urlparse.urljoin(self.hls_url, mp) 
108          else: 
109              self.hls_url = self.getUrl() 
110   
111          self.hlsring.setHostname(self.hls_url) 
112          self.soft_restart() 
113   
114 -    def soft_restart(self): 
115          """Stops serving fragments, resets the playlist and starts 
116          waiting for new segments to become happy again 
117          """ 
118          self.info("Soft restart, resetting playlist and waiting to fill " 
119                    "the initial fragments window") 
120          self._ready = False 
121          self._fragmentsCount = 0 
122          self._last_index = 0 
123          self.hlsring.reset() 
124   
125 -    def _setup_stream_type(self, stream_type): 
126          self.info("Setting up streamer for stream type %s", stream_type) 
127          mime_type, content_type, frag_ext = SUPPORTED_FORMATS[stream_type] 
128          self._mime_type = mime_type 
129          self._content_type = content_type 
130          self.hlsring.filenameExt = frag_ext 
131          self._stream_setup = True 
132   
133 -    def _configure_sink(self): 
134          self.sink.set_property('write-to-disk', False) 
135          self.sink.set_property('playlist-max-window', 5) 
136   
137 -    def _connect_sink_signals(self): 
138          FragmentedStreamer._connect_sink_signals(self) 
139          self.sink.connect("new-fragment", self._new_fragment) 
140   
141 -    def _process_fragment(self, fragment): 
142   
143          if not self._stream_setup: 
144              sink = self.get_element("sink") 
145              pad = sink.get_pad("sink") 
146              caps = pad.get_negotiated_caps() 
147              name = caps.get_structure(0).get_name() 
148              self._setup_stream_type(name) 
149   
150          self._fragmentsCount = self._fragmentsCount + 1 
151   
152          # Wait hls-min-window fragments to set the component 'happy' 
153          if self._fragmentsCount == self._minWindow: 
154              self.info("%d fragments received. Changing mood to 'happy'", 
155                      self._fragmentsCount) 
156              self.setMood(moods.happy) 
157              self._ready = True 
158   
159          b = fragment.get_property('buffer') 
160          index = fragment.get_property('index') 
161          duration = fragment.get_property('duration') 
162   
163          if index < self._last_index: 
164              self.warning("Found a discontinuity last index is %s but current " 
165                           "one is %s", self._last_index, index) 
166              self.soft_restart() 
167   
168          fragName = self.hlsring.addFragment(b.data, index, 
169                  round(duration / float(gst.SECOND))) 
170          self.info('Added fragment "%s", index=%s, duration=%s', 
171                    fragName, index, gst.TIME_ARGS(duration)) 
172   
173      ### START OF THREAD-AWARE CODE (called from non-reactor threads) 
174   
175 -    def _new_fragment(self, hlssink): 
176          self.log("hlsink created a new fragment") 
177          try: 
178              fragment = hlssink.get_property('fragment') 
179          except: 
180              fragment = hlssink.emit('pull-fragment') 
181          reactor.callFromThread(self._process_fragment, fragment) 
182   
183      ### END OF THREAD-AWARE CODE 
184