feat: add use_table_listener hook

plugins/ui/examples/README.md

@@ -595,3 +595,69 @@ st = stock_table(stocks)
 ```
 
 ![Stock Rollup](assets/stock_rollup.png)
+
+## Listening to Table Updates
+
+You can use the `use_table_listener` hook to listen to changes to a table. In this example, we use the `use_table_listener` hook to listen to changes to the table then display the last changes.
+
+```python
+import deephaven.ui as ui
+from deephaven.table import Table
+from deephaven import time_table, empty_table, merge
+from deephaven import pandas as dhpd
+import pandas as pd
+
+
+def to_table(update):
+    return dhpd.to_table(pd.DataFrame.from_dict(update))
+
+
+def add_as_op(ls, t, op):
+    t = t.update(f"type=`{op}`")
+    ls.append(t)
+
+
+@ui.component
+def monitor_changed_data(source: Table):
+
+    changed, set_changed = ui.use_state(empty_table(0))
+
+    show_added, set_show_added = ui.use_state(True)
+    show_removed, set_show_removed = ui.use_state(True)
+
+    def listener(update, is_replay):
+
+        to_merge = []
+
+        if (added_dict := update.added()) and show_added:
+            added = to_table(added_dict)
+            add_as_op(to_merge, added, "added")
+
+        if (removed_dict := update.removed()) and show_removed:
+            removed = to_table(removed_dict)
+            add_as_op(to_merge, removed, "removed")
+
+        if to_merge:
+            set_changed(merge(to_merge))
+        else:
+            set_changed(empty_table(0))
+
+    ui.use_table_listener(source, listener)
+
+    added_check = ui.checkbox(
+        "Show Added", isSelected=show_added, on_change=set_show_added
+    )
+
+    removed_check = ui.checkbox(
+        "Show Removed", isSelected=show_removed, on_change=set_show_removed
+    )
+
+    return [added_check, removed_check, changed]
+
+
+t = time_table("PT1S").update(formulas=["X=i"]).tail(5)
+
+monitor = monitor_changed_data(t)
+```
+
+![Stock Rollup](assets/change_monitor.png)

plugins/ui/src/deephaven/ui/hooks/__init__.py

@@ -3,11 +3,13 @@
 from .use_memo import use_memo
 from .use_state import use_state
 from .use_ref import use_ref
+from .use_table_listener import use_table_listener
 
 __all__ = [
     "use_callback",
     "use_effect",
     "use_memo",
     "use_state",
     "use_ref",
+    "use_table_listener",
 ]

plugins/ui/src/deephaven/ui/hooks/use_table_listener.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from functools import partial
+from typing import Callable, Literal
+
+from deephaven.table import Table
+from deephaven.table_listener import listen, TableUpdate, TableListener
+from deephaven.execution_context import get_exec_ctx, ExecutionContext
+
+from .use_effect import use_effect
+
+LockType: Literal["shared", "exclusive"]
+
+
+def listener_with_ctx(
+    exec_ctx: ExecutionContext,
+    listener: Callable[[TableUpdate, bool], None],
+    update: TableUpdate,
+    is_replay: bool,
+) -> None:
+    """
+    Call the listener within an execution context.
+
+    Args:
+        exec_ctx: ExecutionContext: The execution context to use.
+        listener: Callable[[TableUpdate, bool], None]: The listener to call.
+        update: TableUpdate: The update to pass to the listener.
+        is_replay: bool: Whether the update is a replay.
+    """
+    with exec_ctx:
+        listener(update, is_replay)
+
+
+def with_ctx(
+    listener: Callable[[TableUpdate, bool], None]
+) -> partial[[TableUpdate, bool], None]:
+    """
+    Wrap the listener in an execution context.
+
+    Args:
+        listener: The listener to wrap.
+
+    Returns:
+        partial[[TableUpdate, bool], None]: The wrapped listener.
+    """
+    return partial(listener_with_ctx, get_exec_ctx(), listener)
+
+
+def wrap_listener(
+    listener: Callable[[TableUpdate, bool], None] | TableListener
+) -> partial[[TableUpdate, bool], None]:
+    """
+    Wrap the listener in an execution context.
+
+    Args:
+        listener: Callable[[TableUpdate, bool], None]: The listener to wrap.
+
+    Returns:
+        partial[[TableUpdate, bool], None]: The wrapped listener.
+    """
+    if isinstance(listener, TableListener):
+        return with_ctx(listener.on_update)
+    elif callable(listener):
+        return with_ctx(listener)
+
+
+def use_table_listener(
+    table: Table,
+    listener: Callable[[TableUpdate, bool], None] | TableListener,
+    description: str | None = None,
+    do_replay: bool = False,
+    replay_lock: LockType = "shared",
+) -> None:
+    """
+    Listen to a table and call a listener when the table updates.
+
+    Args:
+        table: Table: The table to listen to.
+        listener: Callable[[TableUpdate, bool], None] | TableListener: Either a function or a TableListener with an
+        on_update function. The function must take a TableUpdate and is_replay bool.
+        description: str | None: An optional description for the UpdatePerformanceTracker to append to the listener’s
+        entry description, default is None.
+        do_replay: bool: Whether to replay the initial snapshot of the table, default is False.
+        replay_lock: LockType: The lock type used during replay, default is ‘shared’, can also be ‘exclusive’.
+    """
+
+    def start_listener() -> Callable[[], None]:
+        """
+        Start the listener. Returns a function that can be called to stop the listener by the use_effect hook.
+
+        Returns:
+            Callable[[], None]: A function that can be called to stop the listener by the use_effect hook.
+        """
+        handle = listen(
+            table,
+            wrap_listener(listener),
+            description=description,
+            do_replay=do_replay,
+            replay_lock=replay_lock,
+        )
+
+        return lambda: handle.stop()
+
+    use_effect(start_listener, [table, listener, description, do_replay, replay_lock])

plugins/ui/test/deephaven/ui/test_hooks.py

@@ -1,5 +1,7 @@
+import threading
 import unittest
 from operator import itemgetter
+from time import sleep
 from typing import Callable
 from unittest.mock import Mock
 from .BaseTest import BaseTestCase
@@ -133,6 +135,33 @@ def _test_memo(fn=lambda: "foo", a=1, b=2):
         self.assertEqual(result, "biz")
         self.assertEqual(mock.call_count, 1)
 
+    def test_table_listener(self):
+        from deephaven.ui.hooks import use_table_listener
+        from deephaven import time_table, new_table, DynamicTableWriter
+        from deephaven.table_listener import TableUpdate
+        import deephaven.dtypes as dht
+
+        column_definitions = {"Numbers": dht.int32, "Words": dht.string}
+
+        table_writer = DynamicTableWriter(column_definitions)
+        table = table_writer.table
+
+        event = threading.Event()
+
+        def listener(update: TableUpdate, is_replay: bool) -> None:
+            nonlocal event
+            event.set()
+
+        def _test_table_listener(replayed_table_val=table, listener_val=listener):
+            use_table_listener(replayed_table_val, listener_val)
+
+        render_hook(_test_table_listener)
+
+        table_writer.write_row(1, "Testing")
+
+        if not event.wait(timeout=1.0):
+            assert False, "listener was not called"
+
 
 if __name__ == "__main__":
     unittest.main()