#669:

* we verify that the pipeline does start within a reasonable amount of time, and if not, we stop the process
* add some new env vars to make it easier to test failures and crashes
* and an exit generic signal so we can tell the other end to stop sending / receiving sound if the sound process died without cleaning up properly
* log when the sound sink process dies (client side only for now)

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

src/xpra/client/ui_client_base.py

@@ -1681,6 +1681,21 @@ def restart():
         soundlog("sound_sink_overrun() will restart in %ims (server supports eos sequence: %s)", delay, self.server_sound_eos_sequence)
         self.timeout_add(delay, restart)
 
+    def sound_sink_exit(self, sound_sink, *args):
+        log("sound_sink_exit(%s, %s) sound_sink=%s", sound_sink, args, self.sound_sink)
+        ss = self.sound_sink
+        if sound_sink!=ss:
+            soundlog("sound_sink_overrun() not the current sink, ignoring it")
+            return
+        if ss and ss.codec:
+            #the mandatory "I've been naughty warning":
+            #we use the "codec" field as guard to ensure we only print this warning once..
+            log.warn("the %s sound sink has stopped", ss.codec)
+            ss.codec = ""
+        #if we had an overrun, we should have restarted things already
+        #(and the guard at the top ensures we don't end up stopping the new sink)
+        self.stop_receiving_sound()
+
     def start_sound_sink(self, codec):
         soundlog("start_sound_sink(%s)", codec)
         assert self.sound_sink is None, "sound sink already exists!"
@@ -1694,6 +1709,7 @@ def start_sound_sink(self, codec):
             ss.connect("state-changed", self.sound_sink_state_changed)
             ss.connect("error", self.sound_sink_error)
             ss.connect("overrun", self.sound_sink_overrun)
+            ss.connect("exit", self.sound_sink_exit)
             from xpra.net.protocol import Protocol
             ss.connect(Protocol.CONNECTION_LOST, self.sound_process_stopped)
             ss.start()

src/xpra/net/subprocess_wrapper.py

@@ -214,7 +214,8 @@ def process_packet(self, proto, packet):
             return
         method = getattr(self.wrapped_object, attr, None)
         if not method:
-            log.warn("unknown command: %s", command)
+            if self.wrapped_object is not None:
+                log.warn("unknown command: %s", command)
             return
         if DEBUG_WRAPPER:
             log("calling %s.%s%s", self.wrapped_object, attr, str(tuple(packet[1:]))[:128])
@@ -249,7 +250,9 @@ def connect(self, signal, cb, *args):
 
 
     def subprocess_exit(self, *args):
+        #beware: this may fire more than once!
         log("subprocess_exit%s command=%s", args, self.command)
+        self._fire_callback("exit")
 
     def start(self):
         self.process = self.exec_subprocess()
@@ -350,13 +353,16 @@ def send(self, *packet_data):
     def process_packet(self, proto, packet):
         if DEBUG_WRAPPER:
             log("process_packet(%s, %s)", proto, [str(x)[:32] for x in packet])
-        command = bytestostr(packet[0])
-        callbacks = self.signal_callbacks.get(command)
-        log("process_packet callbacks(%s)=%s", command, callbacks)
+        signal_name = bytestostr(packet[0])
+        self._fire_callback(signal_name, packet[1:])
+
+    def _fire_callback(self, signal_name, extra_args=[]):
+        callbacks = self.signal_callbacks.get(signal_name)
+        log("firing callback for %s: %s", signal_name, callbacks)
         if callbacks:
             for cb, args in callbacks:
                 try:
-                    all_args = list(packet[1:]) + args
+                    all_args = list(args) + extra_args
                     cb(self, *all_args)
                 except Exception:
-                    log.error("error processing callback %s for %s packet", cb, command, exc_info=True)
+                    log.error("error processing callback %s for %s packet", cb, signal_name, exc_info=True)

src/xpra/server/source.py

@@ -782,10 +782,15 @@ def start_sending_sound(self, codec, volume=1.0):
                 ss.connect("new-buffer", self.new_sound_buffer)
                 ss.connect("new-stream", self.new_stream)
                 ss.connect("info", self.sound_source_info)
+                ss.connect("exit", self.sound_source_exit)
                 ss.start()
         except Exception as e:
             log.error("error setting up sound: %s", e, exc_info=True)
 
+    def sound_source_exit(self, source, *args):
+        log("sound_source_exit(%s, %s)", source, args)
+        self.stop_sending_sound()
+
     def sound_source_info(self, source, info):
         soundlog("sound_source_info: %s", info)
 

src/xpra/sound/wrapper.py

@@ -14,7 +14,10 @@
 
 DEBUG_SOUND = os.environ.get("XPRA_SOUND_DEBUG", "0")=="1"
 SUBPROCESS_DEBUG = os.environ.get("XPRA_SOUND_SUBPROCESS_DEBUG", "").split(",")
-FAKE_OVERRUN = int(os.environ.get("XPRA_FAKE_OVERRUN", "0"))
+FAKE_OVERRUN = int(os.environ.get("XPRA_SOUND_FAKE_OVERRUN", "0"))
+FAKE_START_FAILURE = os.environ.get("XPRA_SOUND_FAKE_START_FAILURE", "0")=="1"
+FAKE_EXIT = int(os.environ.get("XPRA_SOUND_FAKE_EXIT", "0"))
+FAKE_CRASH = int(os.environ.get("XPRA_SOUND_FAKE_CRASH", "0"))
 
 
 #this wrapper takes care of launching src.py or sink.py
@@ -51,7 +54,18 @@ def __init__(self, wrapped_object, method_whitelist, exports_list):
             self.connect_export(x)
 
     def start(self):
-        gobject.idle_add(self.wrapped_object.start)
+        if not FAKE_START_FAILURE:
+            gobject.idle_add(self.wrapped_object.start)
+        if FAKE_EXIT>0:
+            def process_exit():
+                self.cleanup()
+                gobject.timeout_add(250, self.stop)
+            gobject.timeout_add(FAKE_EXIT*1000, process_exit)
+        if FAKE_CRASH>0:
+            def force_exit():
+                import sys
+                sys.exit(1)
+            gobject.timeout_add(FAKE_CRASH*1000, force_exit)
         subprocess_callee.start(self)
 
     def cleanup(self):
@@ -149,6 +163,11 @@ def __init__(self, description):
         self.connect("info", self.info_update)
         self.connect("signal", self.subprocess_signal)
 
+    def start(self):
+        self.verify_started()
+        subprocess_caller.start(self)
+        gobject.timeout_add(2500, self.verify_started)
+
 
     def cleanup(self):
         log("cleanup() sending cleanup request to %s", self.description)
@@ -157,6 +176,18 @@ def cleanup(self):
         gobject.timeout_add(500, self.stop)
 
 
+    def verify_started(self):
+        p = self.process
+        log("verify_started() process=%s, last_info=%s, codec=%s", p, self.last_info, self.codec)
+        if p is None or p.poll() is not None:
+            #process has terminated already
+            return
+        #if we don't get an "info" packet, then the pipeline must have failed to start
+        if not self.last_info:
+            log.warn("the %s process has failed to start, stopping it", self.description)
+            self.cleanup()
+
+
     def subprocess_signal(self, wrapper, proc):
         log("subprocess_signal: %s", proc)
         self.stop_protocol()