gstreamer 1.0 sound support for py3k + win32 clients:

* ensure the codec name is a string
* define more constants and aliases for old function names
* modify add_data / push_buffer so we don't need to access the data inside the buffer (requires different code with each version)
* add more pipeline message handlers
* duplicate all the preroll / buffer grabbing methods since the structures are sufficiently different

git-svn-id: https://xpra.org/svn/Xpra/trunk@6315 3bb7dfac-3a0b-4e04-842a-767bc560f471

src/setup.py

@@ -932,6 +932,7 @@ def add_data_files(target_dir, files):
                             'libsoup-2.4-1.dll',
                             'libvisual-0.4-0.dll',
                             'libgstreamer-1.0-0.dll',
+                            'liborc-test-0.4-0.dll',
                             'libopenjpeg-1.dll',
                             'libsqlite3-0.dll']
                 for p in ("app", "audio", "base", "codecparsers", "fft", "net", "video",

src/xpra/client/ui_client_base.py

@@ -1535,8 +1535,10 @@ def new_sound_buffer(self, sound_source, data, metadata):
 
     def _process_sound_data(self, packet):
         codec, data, metadata = packet[1:4]
+        codec = bytestostr(codec)
+        metadata = typedict(metadata)
         if not self.speaker_enabled:
-            if metadata.get("start-of-stream"):
+            if metadata.boolget("start-of-stream"):
                 #server is asking us to start playing sound
                 if not self.speaker_allowed:
                     #no can do!
@@ -1545,18 +1547,18 @@ def _process_sound_data(self, packet):
                 self.speaker_enabled = True
                 self.emit("speaker-changed")
                 self.on_sink_ready = None
-                codec = metadata.get("codec")
+                codec = metadata.strget("codec")
                 soundlog("starting speaker on server request using codec %s", codec)
                 self.start_sound_sink(codec)
             else:
                 soundlog("speaker is now disabled - dropping packet")
                 return
-        elif metadata.get("end-of-stream"):
+        elif metadata.boolget("end-of-stream"):
             if self.sound_sink:
                 soundlog("server sent end-of-stream, closing sound pipeline")
                 self.stop_receiving_sound(False)
             return
-        seq = metadata.get("sequence", -1)
+        seq = metadata.intget("sequence", -1)
         if self.min_sound_sequence>0 and seq<self.min_sound_sequence:
             soundlog("ignoring sound data with old sequence number %s", seq)
             return

src/xpra/sound/gstreamer_util.py

@@ -92,16 +92,29 @@ def import_gst1():
     Gst.registry_get_default = Gst.Registry.get 
     Gst.get_pygst_version = lambda: gi.version_info
     Gst.get_gst_version = lambda: Gst.version()
-    Gst.new_buffer = Gst.Buffer.new_allocate
+    def new_buffer(data):
+        buf = Gst.Buffer.new_allocate(None, len(data), None)
+        buf.fill(0, data)
+        return buf
+    Gst.new_buffer = new_buffer
+    Gst.element_state_get_name = Gst.Element.state_get_name
     #note: we only copy the constants we actually need..
     for x in ('NULL', 'PAUSED', 'PLAYING', 'READY', 'VOID_PENDING'):
         setattr(Gst, "STATE_%s" % x, getattr(Gst.State, x))
     for x in ('EOS', 'ERROR', 'TAG', 'STREAM_STATUS', 'STATE_CHANGED',
-              'LATENCY', 'WARNING', 'ASYNC_DONE', 'NEW_CLOCK', 'DURATION'):
+              'LATENCY', 'WARNING', 'ASYNC_DONE', 'NEW_CLOCK', 'STREAM_STATUS',
+              'BUFFERING', 'INFO', 'STREAM_START',  
+              ):
         setattr(Gst, "MESSAGE_%s" % x, getattr(Gst.MessageType, x))
+    Gst.MESSAGE_DURATION = Gst.MessageType.DURATION_CHANGED
+    Gst.FLOW_OK = Gst.FlowReturn.OK
+    global gst_version, pygst_version
+    gst_version = Gst.get_gst_version()
+    pygst_version = Gst.get_pygst_version()
     return Gst
 
 def import_gst0_10():
+    global gst_version, pygst_version
     import pygst
     pygst.require("0.10")
     try:
@@ -353,6 +366,9 @@ def main():
             log.enable_debug()
         print("GStreamer plugins found: %s" % ", ".join(get_all_plugin_names()))
         print("")
+        print("GStreamer version: %s" % ".".join([str(x) for x in gst_version]))
+        print("PyGStreamer version: %s" % ".".join([str(x) for x in pygst_version]))
+        print("")
         encs = [x for x in CODEC_ORDER if has_encoder(x)]
         decs = [x for x in CODEC_ORDER if has_decoder(x)]
         print("encoders supported: %s" % str(encs))

src/xpra/sound/sink.py

@@ -160,7 +160,11 @@ def add_data(self, data, metadata=None):
             if d is not None:
                 buf.duration = d
         log("add_data(..) queue_state=%s", self.queue_state)
-        self.push_buffer(buf)
+        if self.push_buffer(buf):
+            self.buffer_count += 1
+            self.byte_count += len(data)
+            ltime = int(self.queue.get_property("current-level-time")/MS_TO_NS)
+            log("sound sink: pushed %s bytes, new buffer level: %sms", len(data), ltime)
 
     def push_buffer(self, buf):
         #buf.size = size
@@ -174,10 +178,6 @@ def push_buffer(self, buf):
             log.error("push-buffer error: %s", r)
             self.emit('error', "push-buffer error: %s" % r)
             return False
-        self.buffer_count += 1
-        self.byte_count += len(buf.data)
-        ltime = int(self.queue.get_property("current-level-time")/MS_TO_NS)
-        log("sound sink: pushed %s bytes, new buffer level: %sms", len(buf.data), ltime)
         return True
 
 gobject.type_register(SoundSink)

src/xpra/sound/sound_pipeline.py

@@ -106,6 +106,12 @@ def cleanup(self):
     def on_message(self, bus, message):
         #log("on_message(%s, %s)", bus, message)
         t = message.type
+        try:
+            #Gst 1.0:
+            structure = message.get_structure()
+        except:
+            #Gst 0.10:
+            structure = message.structure
         if t == gst.MESSAGE_EOS:
             self.pipeline.set_state(gst.STATE_NULL)
             log.info("sound source EOS")
@@ -118,34 +124,36 @@ def on_message(self, bus, message):
             self.state = "error"
             self.emit("state-changed", self.state)
         elif t == gst.MESSAGE_TAG:
-            if message.structure.has_field("bitrate"):
-                new_bitrate = int(message.structure["bitrate"])
+            if structure.has_field("bitrate"):
+                new_bitrate = int(structure["bitrate"])
                 self.update_bitrate(new_bitrate)
-            elif message.structure.has_field("codec"):
-                desc = message.structure["codec"]
+            elif structure.has_field("codec"):
+                desc = structure["codec"]
                 if self.codec_description!=desc:
                     log.info("codec: %s", desc)
                     self.codec_description = desc
-            elif message.structure.has_field("audio-codec"):
-                desc = message.structure["audio-codec"]
+            elif structure.has_field("audio-codec"):
+                desc = structure["audio-codec"]
                 if self.codec_description!=desc:
                     log.info("using audio codec: %s", desc)
                     self.codec_description = desc
-            elif message.structure.has_field("mode"):
-                mode = message.structure["mode"]
+            elif structure.has_field("mode"):
+                mode = structure["mode"]
                 if self.codec_mode!=mode:
                     log("mode: %s", mode)
                     self.codec_mode = mode
             else:
-                #these we just log them:
+                #these, we know about, so we just log them:
                 for x in ("minimum-bitrate", "maximum-bitrate", "channel-mode"):
-                    if message.structure.has_field(x):
-                        v = message.structure[x]
+                    if structure.has_field(x):
+                        v = structure[x]
                         log("tag message: %s = %s", x, v)
                         return      #handled
-                log.info("unknown tag message: %s", message)
+                log.info("unknown sound pipeline tag message: %r", structure)
         elif t == gst.MESSAGE_STREAM_STATUS:
             log("stream status: %s", message)
+        elif t == gst.MESSAGE_STREAM_START:
+            log("stream start: %s", message)
         elif t in (gst.MESSAGE_LATENCY, gst.MESSAGE_ASYNC_DONE, gst.MESSAGE_NEW_CLOCK):
             log("%s", message)
         elif t == gst.MESSAGE_STATE_CHANGED:
@@ -161,9 +169,11 @@ def on_message(self, bus, message):
             log("duration changed: %s", d)
         elif t == gst.MESSAGE_LATENCY:
             log.info("Latency message from %s: %s", message.src, message)
+        elif t == gst.MESSAGE_INFO:
+            log.info("Sound pipeline message: %s", message)
         elif t == gst.MESSAGE_WARNING:
             w = message.parse_warning()
             log.warn("pipeline warning: %s", w[0].message)
             log.info("pipeline warning: %s", w[1:])
         else:
-            log.info("unhandled bus message type %s: %s / %s", t, message, message.structure)
+            log.info("unhandled bus message type %s: %s / %s", t, message, structure)

src/xpra/sound/src.py

@@ -71,8 +71,14 @@ def __init__(self, src_type=DEFAULT_SRC, src_options={}, codec=MP3, volume=1.0,
         self.sink.set_property("drop", False)
         self.sink.set_property("sync", True)
         self.sink.set_property("qos", False)
-        self.sink.connect("new-buffer", self.on_new_buffer)
-        self.sink.connect("new-preroll", self.on_new_preroll)
+        try:
+            #Gst 1.0:
+            self.sink.connect("new-sample", self.on_new_sample)
+            self.sink.connect("new-preroll", self.on_new_preroll1)
+        except:
+            #Gst 0.10:
+            self.sink.connect("new-buffer", self.on_new_buffer)
+            self.sink.connect("new-preroll", self.on_new_preroll0)
 
     def set_volume(self, volume=1.0):
         if self.sink and self.volume:
@@ -88,16 +94,38 @@ def cleanup(self):
         self.src_type = ""
         self.sink = None
 
-    def on_new_preroll(self, appsink):
+
+    def on_new_preroll1(self, appsink):
+        sample = appsink.emit('pull-preroll')
+        log('new preroll1: %s', sample)
+        self.emit_buffer1(sample)
+
+    def on_new_sample(self, bus):
+        #Gst 1.0
+        sample = self.sink.emit("pull-sample")
+        self.emit_buffer1(sample)
+
+    def emit_buffer1(self, sample):
+        buf = sample.get_buffer()
+        #info = sample.get_info()
+        size = buf.get_size()
+        data = buf.extract_dup(0, size)
+        self.do_emit_buffer(data, {"timestamp"  : buf.pts,
+                                   "duration"   : buf.duration})
+
+
+    def on_new_preroll0(self, appsink):
         buf = appsink.emit('pull-preroll')
-        log('new preroll: %s bytes', len(buf))
-        self.emit_buffer(buf)
+        log('new preroll0: %s bytes', len(buf))
+        self.emit_buffer0(buf)
 
     def on_new_buffer(self, bus):
+        #pygst 0.10
         buf = self.sink.emit("pull-buffer")
-        self.emit_buffer(buf)
+        self.emit_buffer0(buf)
+
 
-    def emit_buffer(self, buf):
+    def emit_buffer0(self, buf, metadata={}):
         """ convert pygst structure into something more generic for the wire """
         #none of the metadata is really needed at present, but it may be in the future:
         #metadata = {"caps"      : buf.get_caps().to_string(),
@@ -106,11 +134,14 @@ def emit_buffer(self, buf):
         #            "duration"  : buf.duration,
         #            "offset"    : buf.offset,
         #            "offset_end": buf.offset_end}
-        metadata = {"timestamp" : buf.timestamp,
-                    "duration"  : buf.duration}
+        self.do_emit_buffer(buf.data, {"timestamp" : buf.timestamp,
+                                       "duration"  : buf.duration})
+
+
+    def do_emit_buffer(self, data, metadata={}):
         self.buffer_count += 1
-        self.byte_count += len(buf.data)
-        self.emit("new-buffer", buf.data, metadata)
+        self.byte_count += len(data)
+        self.emit("new-buffer", data, metadata)
 
 gobject.type_register(SoundSource)