Feature: Keep pulse() connected

[timojuez]

p=pulse() shall not kill the process when pulse disconnects.

It happens from time to time that the user might restart or kill pulse e.g. because he changes the config or so. Everyone who implements some daemon that has a pulse() instance might prefer a Pulse class that reconnects by itself as soon as pulse is available. Is there space somewhere for something like this?

import pulsectl, time, sys
from threading import Thread

class Pulse(pulsectl.Pulse):

    def __init__(self,*args,**xargs):
        super().__init__(*args,connect=False,**xargs)
        self.connect_pulse()

    def connect_pulse(self):
        def connect():
            self.connect()
            self.on_connected()
        def keep_reconnecting():
            while True:
                try: connect()
                except pulsectl.pulsectl.PulseError: time.sleep(3)
                else: break
        print("[%s] Connecting..."%self.__class__.__name__, file=sys.stderr)
        try: connect()
        except pulsectl.pulsectl.PulseError:
            Thread(target=keep_reconnecting,daemon=True).start()

    def on_connected(self):
        print("[%s] Connected to Pulseaudio."%self.__class__.__name__, file=sys.stderr)

Of course, using this makes sense with try-catch around all pulse() attribute calls.

[mk-fg]

Under the hood, libpulse runs glib eventloop in one thread, and not sure how well it'd work with calls from another thread. Blocking calls from another thread when loop is running under the hood should definitely cause issues, if not straight-up segfaults, pretty sure.

Some simple loop like this should work though, I think:

pulse = None
try:
  while True:
    pulse = Pulse()
    try: ...anything
    except PulseError:
      if pulse.connected: raise
finally:
  if pulse: pulse.close()

(can probably be wrapped into context abstraction or something nice-ish like that)

[mk-fg]

Oh, also wrt this bit:

print("[%s] Connecting..."%self.__class__.__name__, file=sys.stderr)
try: connect()
except pulsectl.pulsectl.PulseError:
  Thread(target=keep_reconnecting,daemon=True).start()

This should already be implemented by libpulse itself, but I think it's disabled by default, so with loop code above, also do:

pulse = Pulse(connect=False)
pulse.connect(wait=True)

(under the hood wait passes PA_CONTEXT_NOFAIL to context_connect call, which should make it retry connection - see pulseaudio docs on that flag for exact details)

[mk-fg]

And after correction above, I think I've realized that you only wanted to retry initial connection, despite somewhat misleading "keep pulse connected" title, and if that's the case, two lines in previous comment should do the trick, I think.

[mk-fg]

Also maybe avoid subclassing Pulse class if possible - as also mentioned in #42 - it's not really intended for that, but of course can be abused in whatever ways, if more convenient for some specific use-case :)

[timojuez]

Yes, it should just reconnect to Pulse whenever possible and not bother in other cases. A (sub)class that does this would be nice.

pulse = Pulse(connect=False)
pulse.connect(wait=True)

Looks good, but pulse.connect(wait=True) raises PulseError immediately if pulse is not available. Is this a new issue?

[mk-fg]

Looks good, but pulse.connect(wait=True) raises PulseError immediately if pulse is not available. Is this a new issue?

Hm, no, weird, lemme check, maybe I misremember how that flag should work...

[mk-fg]

Yes, it should just reconnect to Pulse whenever possible and not bother in other cases.

Not actually sure what you mean in the second part of that sentence, but guess "Yes" means that I eventually understood intended purpose correctly.

[timojuez]

You understood correctly ;) But the second part means that Pulse() shall not fail nor block on init if it's not possible to connect. That's what I do with threading in the example

[mk-fg]

Oh, right, sorry, didn't get that non-blocking part at all, hmm, don't think that "wait" flag will work for that.

[timojuez]

If that wait flag worked like you said, i could replace the keep_reconnecting function by one line. In case that's possible, leave a note :)

[mk-fg]

Description for this flag is:

    PA_CONTEXT_NOFAIL = 0x0002U
    /**< Don't fail if the daemon is not available when pa_context_connect() is
     * called, instead enter PA_CONTEXT_CONNECTING state and wait for the daemon
     * to appear.  \since 0.9.15 */

And it definitely worked for me before, but now it does not, wonder if might be an upstream bug or change.

[mk-fg]

Implementation for this flag seem to be:

#ifdef HAVE_DBUS
            if (c->no_fail && !c->server_specified) {
                if (!c->session_bus)
                    track_pulseaudio_on_dbus(c, DBUS_BUS_SESSION, &c->session_bus);
                if (!c->system_bus)
                    track_pulseaudio_on_dbus(c, DBUS_BUS_SYSTEM, &c->system_bus);

                if (c->session_bus || c->system_bus) {
                    pa_log_debug("Waiting for PA on D-Bus...");
                    break;
                }
            } else
#endif
                pa_context_fail(c, PA_ERR_CONNECTIONREFUSED);

So I think track_pulseaudio_on_dbus() might've stopped working at some point, or maybe pulse is built without dbus, not sure if it matters which case it is, flag seem to be generally unreliable in any case - it's not just reconnect-loop, and is blocking anyway.

Don't think there's much point adding async connection operation here, as that seem to be kinda against what it does (blocking synchronous API) and would probably require threading magic, locks and such, which is probably best left for whatever integration code.

Apologies for confusion wrt what you were asking above.

Still not sure why you need subclass and threads in the code example above, but of course I don't know full use-case for it, so not really supposed to anyway :) Will look at making it easier for subclassing in #42 though.

[timojuez]

Ok, fine. The reason why I use the asynchronous connection is this: It happens from time to time that the user might restart or kill pulse e.g. because he changes the config or so. If this happens while my program runs, it crashes

[mk-fg]

Right, I think you mentioned this exact reason at the top as well, and I guess async is there because - as you also said above - this connection routine can't be blocking main app. So yeah, makes sense I guess, and threads are probably reasonable way to wrap this for a blocking module like this one.

Integrating libpulse into main app's eventloop might be a way to avoid threads here, but that'd need a non-blocking module (e.g. something asyncio-based or glib-based gi bindings) and not this one, plus maybe more trouble/complexity than it's worth (depending on main app's eventloop).

Thanks for explaining, I'm just a bit slow :(