Add on_clear handlers to (Global)Table and ChangeloggedSet (#645)

* #601 - Adding on_clear handlers to Table and ChangeloggedSet

* Fix formatting

* Fix formatting

Author: egao1980

faust/tables/sets.py

@@ -84,6 +84,11 @@ def on_discard(self, value: VT) -> None:
     def on_change(self, added: Set[VT], removed: Set[VT]) -> None:
         self.manager.send_changelog_event(self.key, OPERATION_UPDATE, [added, removed])
 
+    def on_clear(self) -> None:
+        self.manager.send_changelog_event(
+            self.key, OPERATION_UPDATE, [set(), set(self.data)]
+        )
+
     def sync_from_storage(self, value: Any) -> None:
         self.data = set(value)
 

faust/tables/table.py

@@ -93,6 +93,11 @@ def on_key_del(self, key: KT) -> None:
         self._maybe_del_key_ttl(key, partition)
         self._sensor_on_del(self, key)
 
+    def on_clear(self) -> None:
+        """Call when the table is cleared."""
+        for key in self:
+            self.on_key_del(key)
+
     def as_ansitable(self, title: str = "{table.name}", **kwargs: Any) -> str:
-        """Draw table as a a terminal ANSI table."""
+        """Draw table as a terminal ANSI table."""
         return dict_as_ansitable(self, title=title.format(table=self), **kwargs)

tests/unit/tables/test_sets.py

@@ -105,6 +105,15 @@ def test_on_discard(self, *, cset, manager, key):
             "value",
         )
 
+    def test_on_clear(self, *, cset, manager, key):
+        cset.data.add("value")
+        cset.clear()
+        manager.send_changelog_event.assert_called_once_with(
+            key,
+            OPERATION_UPDATE,
+            [set(), {"value"}],
+        )
+
     def test_on_change__diff(self, *, cset, manager, key):
         cset.data.update({1, 2, 3, 4})
         cset.difference_update({2, 3, 4, 5, 6})

tests/unit/tables/test_table.py

@@ -139,6 +139,18 @@ def test_del_key(self, *, table):
             table.send_changelog.asssert_called_once_with(partition, "foo", None)
             assert "foo" not in table.data
 
+    def test_clear(self, *, table):
+        with patch("faust.tables.base.current_event") as current_event:
+            event = current_event.return_value
+            partition = event.message.partition
+            table.send_changelog = Mock(name="send_changelog")
+            table.data["foo"] = 3
+            table.data["bar"] = 4
+            table.clear()
+            table.send_changelog.asssert_called_once_with(partition, "foo", None)
+            table.send_changelog.asssert_called_once_with(partition, "bar", None)
+            assert not table.data
+
     def test_as_ansitable(self, *, table):
         table.data["foo"] = "bar"
         table.data["bar"] = "baz"