diff --git a/nion/utils/Stream.py b/nion/utils/Stream.py index adddb51..f57d603 100644 --- a/nion/utils/Stream.py +++ b/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.""" diff --git a/nion/utils/test/Stream_test.py b/nion/utils/test/Stream_test.py index adcd84f..1615286 100644 --- a/nion/utils/test/Stream_test.py +++ b/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)