Rename FollowStream to RelayStream. Add AsyncRelayStream.

nion/utils/Stream.py

@@ -403,31 +403,64 @@ def __value_changed(self, value: typing.Optional[T]) -> None:
             self.value = None
 
 
-class FollowStream(ValueStream[T], typing.Generic[T]):
+class RelayStream(ValueStream[T], typing.Generic[T]):
     """Sends value from input stream. Input stream can be changed."""
 
-    def __init__(self, stream: typing.Optional[AbstractStream[T]] = None) -> None:
+    def __init__(self, stream: AbstractStream[T] | None = None) -> None:
         super().__init__()
         self.__stream: AbstractStream[T]
         self.__stream_action: ValueStreamAction[T]
         self.stream = stream
 
     @property
-    def stream(self) -> typing.Optional[AbstractStream[T]]:
+    def stream(self) -> AbstractStream[T] | None:
         return self.__stream
 
     @stream.setter
-    def stream(self, stream: typing.Optional[AbstractStream[T]]) -> None:
+    def stream(self, stream: AbstractStream[T] | None) -> None:
         stream = stream or ConstantStream[T](None)
         self.__stream = stream
-        self.__stream_action = ValueStreamAction(self.__stream, weak_partial(FollowStream.__value_changed, self))
-        self.__value_changed(self.__stream.value)
+        self.__stream_action = ValueStreamAction(self.__stream, weak_partial(self.__class__._source_value_changed, self))
+        self._dispatch_value(self.__stream.value, True)
 
-    # define a stub and use weak_partial to avoid holding references to self.
-    def __value_changed(self, value: typing.Optional[T]) -> None:
+    def _source_value_changed(self, value: T | None) -> None:
+        self._dispatch_value(value, False)
+
+    def _dispatch_value(self, value: T | None, is_initial: bool) -> None:
+        # Override to customize where/when values are delivered.
+        # `is_initial` is True for the immediate value sync when `stream` is initialized.
+        self._value_changed(value)
+
+    def _value_changed(self, value: T | None) -> None:
+        # Override to customize how this relay stores/emits a delivered value.
+        # The default writes directly to this stream's `value` property.
         self.value = value
 
 
+# Deprecated alias, kept for backward compatibility.
+FollowStream = RelayStream
+
+
+class AsyncRelayStream(RelayStream[T], typing.Generic[T]):
+    """A stream that relays values from an input stream but sends values via an event loop.
+
+    This can be used to transfer stream values received from another thread to the main thread. Useful for UI.
+    """
+
+    def __init__(self, event_loop: asyncio.AbstractEventLoop, stream: AbstractStream[T] | None = None) -> None:
+        self.__event_loop = event_loop
+        super().__init__(stream)
+
+    def _dispatch_value(self, value: T | None, is_initial: bool) -> None:
+        if is_initial:
+            self.value = value
+            return
+
+        def send_value(stream: AsyncRelayStream[T], value: T | None) -> None:
+            stream.value = value
+
+        self.__event_loop.call_soon_threadsafe(weak_partial(send_value, self, value))
+
 
 class PrintStream:
     """Prints value from input stream."""

nion/utils/test/Stream_test.py

@@ -2,6 +2,7 @@
 import asyncio
 import contextlib
 import logging
+import threading
 import typing
 import unittest
 import weakref
@@ -75,6 +76,32 @@ def test_refcounts(self) -> None:
             del stream7
             self.assertIsNone(stream_ref7())
 
+    def test_async_relay_stream_from_other_thread(self) -> None:
+        event_loop = asyncio.new_event_loop()
+        input_stream = Stream.ValueStream[int](0)
+        relay_stream = Stream.AsyncRelayStream[int](event_loop, input_stream)
+        def update_relay_stream() -> None:
+            input_stream.value = 42
+        thread = threading.Thread(target=update_relay_stream)
+        thread.start()
+        thread.join()
+        self.assertEqual(0, relay_stream.value)
+        event_loop.stop()
+        event_loop.run_forever()
+        event_loop.close()
+        self.assertEqual(42, relay_stream.value)
+
+    def test_async_relay_stream_single_thread(self) -> None:
+        event_loop = asyncio.new_event_loop()
+        input_stream = Stream.ValueStream[int](0)
+        relay_stream = Stream.AsyncRelayStream[int](event_loop, input_stream)
+        input_stream.value = 42
+        self.assertEqual(0, relay_stream.value)
+        event_loop.stop()
+        event_loop.run_forever()
+        event_loop.close()
+        self.assertEqual(42, relay_stream.value)
+
 
 if __name__ == '__main__':
     logging.getLogger().setLevel(logging.DEBUG)