Implement dbus remove match

pkg/base1/test-dbus-common.js

@@ -284,13 +284,15 @@ export function common_dbus_tests(channel_options, bus_name) { // eslint-disable
 
     QUnit.test("signal unsubscribe", async assert => {
         let received = false;
+        let signals = 0;
         const dbus = cockpit.dbus(bus_name, channel_options);
 
         function on_signal() {
             received = true;
+            signals++;
         }
 
-        const subscription = dbus.subscribe({
+        let subscription = dbus.subscribe({
             interface: "com.redhat.Cockpit.DBusTests.Frobber",
             path: "/otree/frobber"
         }, on_signal);
@@ -302,6 +304,19 @@ export function common_dbus_tests(channel_options, bus_name) { // eslint-disable
         received = false;
         await dbus.call("/otree/frobber", "com.redhat.Cockpit.DBusTests.Frobber", "RequestSignalEmission", [0]);
         assert.equal(received, false, "signal not received");
+        assert.equal(signals, 1, "received exactly one signal");
+
+        subscription = dbus.subscribe({
+            interface: "com.redhat.Cockpit.DBusTests.Frobber",
+            path: "/otree/frobber"
+        }, on_signal);
+        await dbus.call("/otree/frobber", "com.redhat.Cockpit.DBusTests.Frobber", "RequestSignalEmission", [0]);
+        assert.equal(received, true, "signal received");
+        assert.equal(signals, 2, "received exactly two signals, watch was removed");
+        subscription.remove();
+        await dbus.call("/otree/frobber", "com.redhat.Cockpit.DBusTests.Frobber", "RequestSignalEmission", [0]);
+        assert.equal(signals, 2, "received exactly three signals, ");
+        dbus.close();
     });
 
     QUnit.test("with types", assert => {

src/cockpit/channels/dbus.py

@@ -266,7 +266,7 @@ async def get_ready() -> None:
         else:
             self.ready()
 
-    def add_signal_handler(self, handler, **kwargs):
+    def get_r(self, **kwargs):
         r = dict(**kwargs)
         r['type'] = 'signal'
         if 'sender' not in r and self.name is not None:
@@ -276,6 +276,14 @@ def add_signal_handler(self, handler, **kwargs):
         if r.get('path_namespace') == "/":
             del r['path_namespace']
 
+        return r
+
+    def get_r_string(self, r):
+        return ','.join(f"{key}='{value}'" for key, value in r.items())
+
+    def add_signal_handler(self, handler, **kwargs):
+        r = self.get_r(**kwargs)
+
         def filter_owner(message):
             if self.owner is not None and self.owner == message.get_sender():
                 handler(message)
@@ -285,6 +293,7 @@ def filter_owner(message):
         else:
             func = handler
         r_string = ','.join(f"{key}='{value}'" for key, value in r.items())
+        logger.debug("r_string %s", r_string)
         if not self.is_closing():
             # this gets an EINTR very often especially on RHEL 8
             while True:
@@ -294,7 +303,7 @@ def filter_owner(message):
                 except InterruptedError:
                     pass
 
-            self.matches.append(match)
+            self.matches.append((r_string, match))
 
     def add_async_signal_handler(self, handler, **kwargs):
         def sync_handler(message):
@@ -363,7 +372,7 @@ async def do_add_match(self, message):
         logger.debug('adding match %s', add_match)
 
         async def match_hit(message):
-            logger.debug('got match')
+            logger.debug('got match %s', message)
             async with self.watch_processing_lock:
                 self.send_json(signal=[
                     message.get_path(),
@@ -374,6 +383,19 @@ async def match_hit(message):
 
         self.add_async_signal_handler(match_hit, **add_match)
 
+    async def do_remove_match(self, message):
+        remove_match = message['remove-match']
+        logger.debug('remove match %s', remove_match)
+        r = self.get_r(**remove_match)
+        r_string = self.get_r_string(r)
+        for index, (r_key, slot) in enumerate(self.matches):
+            if r_key == r_string:
+                logger.debug('got match %s %d', self.matches, index)
+                slot.cancel()
+                del self.matches[index]
+                logger.debug('removed match %s %d', self.matches, index)
+                break
+
     async def setup_objectmanager_watch(self, path, interface_name, meta, notify):
         # Watch the objects managed by the ObjectManager at "path".
         # Properties are not watched, that is done by setup_path_watch
@@ -505,6 +527,8 @@ def do_data(self, data):
             self.create_task(self.do_call(message))
         elif 'add-match' in message:
             self.create_task(self.do_add_match(message))
+        elif 'remove-match' in message:
+            self.create_task(self.do_remove_match(message))
         elif 'watch' in message:
             self.create_task(self.do_watch(message))
         elif 'meta' in message:
@@ -514,7 +538,7 @@ def do_data(self, data):
             return
 
     def do_close(self):
-        for slot in self.matches:
+        for (_, slot) in self.matches:
             slot.cancel()
         self.matches = []
         self.close()