Repository: mikedevnull/upnp-av-control

Issue:
Add replay feature to observable
Add the ability to replay the N (or at least 1) last values / notifications to a new subscription on an observable.
This way, we could ensure that any subscriber will now the latest state without actively querying, and any missed updates due to a lost frontend/backend connection might be solved with this as well.
Thus, might be a good way to solve #14  as well.

Discussion:

Patch:
diff --git a/upnpavcontrol/core/oberserver.py b/upnpavcontrol/core/oberserver.py
index a626045..5be029f 100644
--- a/upnpavcontrol/core/oberserver.py
+++ b/upnpavcontrol/core/oberserver.py
@@ -13,6 +13,7 @@ class Subscription(object):
 
     Can be used to unsubscribe from notifications later on
     """
+
     def __init__(self, observable):
         self._observable = observable
 
@@ -47,10 +48,14 @@ class Observable(Generic[T]):
     While the callback is invoked, the subscription count is guaranteed not to change.
     The callback must not add or remove subscriptions, as this will cause a deadlock.
     """
-    def __init__(self):
+
+    def __init__(self, replay=False):
         self._subscriptions: Dict[Subscription, Callable[[T], Awaitable[None]]] = {}
         self._lock = asyncio.Lock()
         self._change_callback_cb: Optional[Callable[[int], Awaitable[None]]] = None
+        self._is_replaying = replay
+        self._last_value: Optional[T] = None
+        self._has_last_value = False
 
     @property
     def subscription_count(self):
@@ -78,6 +83,8 @@ class Observable(Generic[T]):
             self._subscriptions[subscription] = subscriber
             if self._change_callback_cb is not None:
                 await self._change_callback_cb(len(self._subscriptions))
+            if self._has_last_value and self._is_replaying:
+                await subscriber(self._last_value)
         return subscription
 
     async def notify(self, payload: T):
@@ -87,6 +94,8 @@ class Observable(Generic[T]):
         subscriptions = []
         tasks = []
         async with self._lock:
+            self._last_value = payload
+            self._has_last_value = True
             for subscription, callable in self._subscriptions.items():
                 subscriptions.append(subscription)
                 tasks.append(callable(payload))
@@ -139,7 +148,7 @@ async def wait_for_value_if(observerable: Observable[T], predicate: Callable[[T]
                 future.set_result(True)
         except Exception as e:
             _logger.exception('predicate')
-            raise e
+            future.set_exception(e)
 
     subscription = await observerable.subscribe(f)
     try:


Test patch:
diff --git a/tests/unit/test_observable.py b/tests/unit/test_observable.py
index 06f1aba..45fd762 100644
--- a/tests/unit/test_observable.py
+++ b/tests/unit/test_observable.py
@@ -1,6 +1,7 @@
 from upnpavcontrol.core import oberserver
 from ..testsupport import AsyncMock
 import pytest
+import typing
 import asyncio
 
 
@@ -86,6 +87,41 @@ async def test_subscription_callback():
     subscription_cb.assert_called_once_with(0)
 
 
+@pytest.mark.asyncio
+async def test_replay_last_value_on_subscription():
+    callback = AsyncMock()
+
+    subscription_cb = AsyncMock()
+    observable = oberserver.Observable[int](replay=True)
+    observable.on_subscription_change = subscription_cb
+    await observable.notify(42)
+
+    await observable.subscribe(callback)
+    subscription_cb.assert_called_once_with(1)
+    callback.assert_called_once_with(42)
+
+    callback.reset_mock()
+
+    await observable.notify(21)
+    callback.assert_called_once_with(21)
+
+
+@pytest.mark.asyncio
+async def test_no_replay_on_subscription_without_initial_value():
+    callback = AsyncMock()
+
+    subscription_cb = AsyncMock()
+    observable = oberserver.Observable[typing.Optional[int]](replay=True)
+    observable.on_subscription_change = subscription_cb
+
+    await observable.subscribe(callback)
+    subscription_cb.assert_called_once_with(1)
+    callback.assert_not_called()
+
+    await observable.notify(None)
+    callback.assert_called_once_with(None)
+
+
 @pytest.mark.asyncio
 async def test_wait_for_with_predicate_resolves_immediately():
     observable = oberserver.Observable[int]()
@@ -111,6 +147,19 @@ async def test_wait_for_with_predicate_resolves_later():
     assert True
 
 
+@pytest.mark.asyncio
+async def test_wait_for_value_reraises_exceptions_from_predicate():
+    observable = oberserver.Observable[int]()
+
+    def throwing_predicate(value):
+        raise RuntimeError('exception raised in predicate')
+
+    with pytest.raises(RuntimeError):
+        async with oberserver.wait_for_value_if(observable, throwing_predicate, 1):
+            await observable.notify(41)
+    assert observable.subscription_count == 0
+
+
 @pytest.mark.asyncio
 async def test_wait_for_with_predicate_times_out():
     observable = oberserver.Observable[int]()