Stopping emission of a signal

The signal mechanism allows unrelated tasks in a program to respond to the same trigger by connecting a handler in each task to the appropriate signal. If a particular task wants to suppress handling of the signal, there is a well documented mechanism using the handler_block method. However, I have encountered situations where I would like for a task to be able to suppress handling of a signal by all tasks connected to a signal – in essence, to suppress emission of the signal. One situation I have encountered is related to selection of a row in a TreeView. When a user clicks on a row, the TreeSelection associated with the TreeView emits a changed signal. It triggers several unrelated actions in my program. However, I also have a situation where I want to set the selection programmatically, in which case nothing else should happen. Here is a program that illustrates the issue:

import gi
gi.require_version('GLib', '2.0')
gi.require_version('GObject', '2.0')
from gi.repository import GLib, GObject

class Sender(GObject.Object):
    @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST)
    def signal(self):
        pass

    def simulate_click_in_sender(self):
        self.emit('signal')

    def programmatic_change_in_sender(self):
        self.emit('signal')

class Client(GObject.Object):
    def __init__(self, sender, index):
        super().__init__()
        self.handler_id = sender.connect('signal', self.on_signal, index)

    def on_signal(self, obj, index):
        print(f'Client with index {index} received signal')

    def make_programmatic_change_to_sender(self):
        sender.programmatic_change_in_sender()


sender = Sender()
clients = [Client(sender, i) for i in range(3)]

events = []
def event():
    print('Clicking a widget triggers a signal.')
    sender.simulate_click_in_sender()
events.append(event)

def event():
    print('\nHowever, a programmatic change also triggers a signal.')
    clients[0].make_programmatic_change_to_sender()
events.append(event)

def event():
    print('\nIf a client knows that it is not interested in signal when '
        'it is triggered by a programmatic change, it can block its own '
        'handler for signal.')
    with sender.handler_block(clients[0].handler_id):
        clients[0].make_programmatic_change_to_sender()
events.append(event)

def event():
    print('\nWhat I want is a way to block emission of the signal so that '
        'no client receives signal.')
    loop.quit()
events.append(event)

for seconds, event in enumerate(events):
    GLib.timeout_add_seconds(seconds, event)

loop = GLib.MainLoop()
loop.run()

In this program, Sender simulates a widget that produces a signal and Client represents the various chunks of code interested in that signal. I use handler_block to prevent client[0] from responding to signal, but what I want is a way for client[0] to prevent all clients from responding.

I thought of modifying Sender so that it suppresses emission when I set a flag:

class Sender(GObject.Object):
    @GObject.Signal(flags=GObject.SignalFlags.RUN_LAST, 
                    return_type=bool,
                    accumulator=GObject.signal_accumulator_true_handled)
    def signal(self):
        return self.do_not_send

    def __init__(self):
        super().__init__()
        self.do_not_send = False

    def simulate_click_in_sender(self):
        self.emit('signal')

    def programmatic_change_in_sender(self):
        self.emit('signal')

and then I added this event:

def event():
    print('\nFirst set do_not_send to True, then make programmatic change.')
    sender.do_not_send = True
    clients[0].make_programmatic_change_to_sender()
events.append(event)

This technique works in this example. However, I have two issues: I must use connect_after instead of connect in Client. I thought that it might be possible to avoid that inconvenience by changing the flag in the signal to RUN_FIRST, but when I make that change I get an error.

The other issue is that I am not sure how I would use this solution in actual code where Sender is replaced by an object from the library. I could subclass TreeSelection so that the subclass intercepts the changed signal and re-emits it as a customized my-changed-signal, but that solution seems clumsy. Is there a better one? It seems as if my situation is not so uncommon, so I feel that I must be missing something obvious.

I found a solution using signal_stop_emission_by_name:

from contextlib import contextmanager

import gi
gi.require_version('GLib', '2.0')
gi.require_version('GObject', '2.0')
gi.require_version('Gtk', '3.0')
from gi.repository import GLib, GObject, Gtk

@contextmanager
def emission_stopper(obj, signal):
    obj._stop_emission = signal
    yield
    del obj._stop_emission

class MyButton(Gtk.Button):
    def __init__(self):
        super().__init__()
        self.set_label('MyButton')
        self.show()

    def do_clicked(self):
        try:
            if self._stop_emission == 'clicked':
                GObject.signal_stop_emission_by_name(self, 'clicked')
        except AttributeError:
            pass

class Client(GObject.Object):
    def __init__(self, button, index):
        super().__init__()
        self.button = button
        button.connect('clicked', self.on_signal, index)

    def on_signal(self, obj, index):
        print(f'Client with index {index} received clicked')

    def programmatic_click(self):
        self.button.clicked()


class Tester(Gtk.Window):
    def __init__(self):
        super().__init__()
        self.set_default_size(100, 100)

        self.button = button = MyButton()

        self.add(button)
        self.show()

        self.connect_after('destroy', self.on_destroy)

    def on_destroy(self, window):
        loop.quit()

tester = Tester()
clients = [Client(tester.button, i) for i in range(3)]

events = []
def event():
    print('\nClick button from clients[0]')
    clients[0].programmatic_click()
events.append(event)

def event():
    print('\nClick button in stopper context')
    with emission_stopper(tester.button, 'clicked'):
        tester.button.clicked()
        clients[0].programmatic_click()
events.append(event)

def event():
    print('\nClick button again without setting set_stop_emission')
    clients[0].programmatic_click()
events.append(event)

for seconds, event in enumerate(events):
    GLib.timeout_add_seconds(seconds, event)

loop = GLib.MainLoop()
loop.run()

If you click on the button, all the clients receive the signal, but when the program clicks the button from within the emission_stopper context, no client receives the signal.

This topic was automatically closed 14 days after the last reply. New replies are no longer allowed.