Add support for async watcher to rx

param/reactive.py

@@ -86,6 +86,7 @@
 import operator
 
 from collections.abc import Iterable, Iterator
+from functools import partial
 from types import FunctionType, MethodType
 from typing import Any, Callable, Optional
 
@@ -363,21 +364,13 @@ def watch(self, fn=None, onlychanged=True, queued=False, precedence=0):
         self._watch(fn, onlychanged=onlychanged, queued=queued, precedence=precedence)
 
     def _watch(self, fn=None, onlychanged=True, queued=False, precedence=0):
-        def cb(*args):
-            # Resolve value to ensure eager evaluation, then apply func if provided
-            ret = self.value
-            if fn is not None:
-                fn(ret)
-
-        if isinstance(self._reactive, rx):
-            params = self._reactive._params
-        else:
-            params = resolve_ref(self._reactive)
-        for _, group in full_groupby(params, lambda x: id(x.owner)):
-            group[0].owner.param._watch(
-                cb, [dep.name for dep in group], onlychanged=onlychanged,
-                queued=queued, precedence=precedence
-            )
+        def cb(value):
+            from .parameterized import async_executor
+            if iscoroutinefunction(fn):
+                async_executor(partial(fn, value))
+            elif fn is not None:
+                fn(value)
+        bind(cb, self._reactive, watch=True)
 
 
 def bind(function, *args, watch=False, **kwargs):

tests/testreactive.py

@@ -387,6 +387,18 @@ def test_reactive_watch_on_set_input():
     string.rx.value = 'new string'
     assert items == ['new string!']
 
+async def test_reactive_watch_async_on_event():
+    p = Parameters()
+    event = p.param.event.rx()
+    items = []
+    event.rx.watch(items.append)
+    p.param.trigger('event')
+    for _ in range(3):
+        await asyncio.sleep(0.05)
+        if items == [True]:
+            break
+    assert items == [True]
+
 def test_reactive_set_value_non_root_raises():
     rx_val = rx(1) + 1
     with pytest.raises(AttributeError):