diff --git a/pyproject.toml b/pyproject.toml
index f01826f..e686e8e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -87,6 +87,7 @@ dev = [
     "harp-data",
     "harp-benchmarks",
     "typing-extensions>=4.15.0",
+    "reactivex>=4",  # for harp-device[rx] adapter + its tests
 ]
 
 [tool.codespell]
diff --git a/scripts/event_monitor.py b/scripts/event_monitor.py
new file mode 100644
index 0000000..e70c609
--- /dev/null
+++ b/scripts/event_monitor.py
@@ -0,0 +1,72 @@
+"""Example: subscribe to all events from a Harp device and print them.
+
+Connects to a device on COM3, registers a single catch-all handler for *every*
+event, and prints each one to the console until you press Ctrl+C.
+
+Run with:
+    uv run python scripts/event_monitor.py
+"""
+
+import time
+
+from harp.device import (
+    REGISTER_MAP,
+    Device,
+    OperationControl,
+    OperationControlPayload,
+    OperationMode,
+    TimestampSeconds,
+)
+from harp.device.rx import observe
+from harp.protocol import HarpMessage, MessageType
+from harp.serial import open_serial_device
+from reactivex import operators as ops
+
+PORT = "COM3"
+
+
+def print_event(msg: HarpMessage) -> None:
+    if msg.address == 44:  # too damn noisy
+        return
+    register = REGISTER_MAP.get(msg.address, None)
+    if register is not None:
+        value = register.parse(msg)
+    else:
+        value = msg.payload.hex()
+    print(f"NonRx: {msg.timestamp:.6f}  {msg.address}  {msg.message_type.name:<5s}  {value}")
+
+
+if __name__ == "__main__":
+    with open_serial_device(Device, port=PORT) as dev:
+        dev.write(
+            OperationControl,
+            OperationControlPayload(
+                operation_mode=OperationMode.ACTIVE,
+                dump_registers=True,
+                heartbeat=True,
+                mute_replies=False,
+                operation_led=True,
+                visual_indicators=True,
+            ),
+        )
+        disposable = (
+            observe(dev, TimestampSeconds)
+            .pipe(
+                ops.filter(lambda msg: msg.message_type == MessageType.Event),
+                ops.filter(lambda msg: msg.parsed % 2 == 0),
+            )
+            .subscribe(
+                lambda msg: print(
+                    f"Rx:{msg.timestamp:.6f}  {msg.address}  {msg.message_type.name:<5s}  {msg.parsed}"
+                )
+            )
+        )
+
+        dev.subscribe_all(print_event)
+        print(f"Listening for events on {PORT}. Press Ctrl+C to stop.\n")
+        try:
+            while True:
+                time.sleep(1.0)
+        except KeyboardInterrupt:
+            print("\nStopping.")
+            disposable.dispose()
diff --git a/src/packages/harp-device/pyproject.toml b/src/packages/harp-device/pyproject.toml
index 3101f3c..b96b7b1 100644
--- a/src/packages/harp-device/pyproject.toml
+++ b/src/packages/harp-device/pyproject.toml
@@ -7,6 +7,9 @@ dependencies = [
     "harp-protocol",
 ]
 
+[project.optional-dependencies]
+rx = ["reactivex>=4"]
+
 [build-system]
 requires = ["uv_build>=0.9.5"]
 build-backend = "uv_build"
diff --git a/src/packages/harp-device/src/harp/device/rx.py b/src/packages/harp-device/src/harp/device/rx.py
new file mode 100644
index 0000000..2ed0586
--- /dev/null
+++ b/src/packages/harp-device/src/harp/device/rx.py
@@ -0,0 +1,69 @@
+from typing import TYPE_CHECKING, Any, TypeVar
+
+try:
+    import reactivex as rx
+    import reactivex.operators as ops
+    from reactivex.disposable import Disposable
+except ModuleNotFoundError as exc:
+    raise ImportError(
+        "harp.device.rx requires the 'reactivex' package, which is not installed. "
+        "Install the optional extra with:  pip install 'harp-device[rx]'."
+    ) from exc
+
+from harp.protocol import HarpMessage, MessageType
+from harp.protocol._message import ParsedHarpMessage
+from harp.protocol._register import RegisterBase
+
+from ._device import MessageTypeFilter
+
+if TYPE_CHECKING:
+    from ._device import Device
+
+P = TypeVar("P")
+
+__all__ = ["observe", "observe_all"]
+
+
+def observe(
+    device: "Device",
+    register: type[RegisterBase[P]],
+    *,
+    message_types: MessageTypeFilter = MessageType.Event,
+) -> "rx.Observable[ParsedHarpMessage[P]]":
+    """Return a hot, multicast Observable of parsed messages for ``register``.
+
+    Mirrors :meth:`Device.subscribe`: by default only ``Event`` messages are
+    emitted; pass ``message_types`` for read/write replies too.
+
+    The stream is ref-counted (``ops.share``): the *first* subscriber creates a
+    single underlying :meth:`Device.subscribe`, additional subscribers share it,
+    and the device subscription is disposed once the last one unsubscribes.
+
+    Note: emissions run on the device's shared event thread. Use
+    ``ops.observe_on(scheduler)`` to move heavy downstream work off it.
+    """
+
+    def on_subscribe(observer: "rx.abc.ObserverBase[Any]", _: Any = None) -> Disposable:
+        sub = device.subscribe(register, observer.on_next, message_types=message_types)
+        return Disposable(sub.unsubscribe)
+
+    return rx.create(on_subscribe).pipe(ops.share())
+
+
+def observe_all(
+    device: "Device",
+    *,
+    message_types: MessageTypeFilter = MessageType.Event,
+) -> "rx.Observable[HarpMessage]":
+    """Return a hot, multicast Observable of raw messages for *every* address.
+
+    Mirrors :meth:`Device.subscribe_all` (a full-traffic firehose when passed
+    more message types). Same ref-counted, hot, single-underlying-subscription
+    lifecycle and threading semantics as :func:`observe`.
+    """
+
+    def on_subscribe(observer: "rx.abc.ObserverBase[Any]", _: Any = None) -> Disposable:
+        sub = device.subscribe_all(observer.on_next, message_types=message_types)
+        return Disposable(sub.unsubscribe)
+
+    return rx.create(on_subscribe).pipe(ops.share())
diff --git a/uv.lock b/uv.lock
index d465ba6..db348e1 100644
--- a/uv.lock
+++ b/uv.lock
@@ -323,6 +323,7 @@ dev = [
     { name = "harp-serial" },
     { name = "pytest" },
     { name = "pytest-cov" },
+    { name = "reactivex" },
     { name = "ruff" },
     { name = "ty" },
     { name = "typing-extensions" },
@@ -355,6 +356,7 @@ dev = [
     { name = "harp-serial", editable = "src/packages/harp-serial" },
     { name = "pytest", specifier = ">=8.3.5" },
     { name = "pytest-cov", specifier = ">=6.1.1" },
+    { name = "reactivex", specifier = ">=4" },
     { name = "ruff", specifier = ">=0.11.0" },
     { name = "ty", specifier = ">=0.0.0" },
     { name = "typing-extensions", specifier = ">=4.15.0" },
@@ -415,8 +417,17 @@ dependencies = [
     { name = "harp-protocol" },
 ]
 
+[package.optional-dependencies]
+rx = [
+    { name = "reactivex" },
+]
+
 [package.metadata]
-requires-dist = [{ name = "harp-protocol", editable = "src/packages/harp-protocol" }]
+requires-dist = [
+    { name = "harp-protocol", editable = "src/packages/harp-protocol" },
+    { name = "reactivex", marker = "extra == 'rx'", specifier = ">=4" },
+]
+provides-extras = ["rx"]
 
 [[package]]
 name = "harp-protocol"
@@ -1114,6 +1125,18 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/04/11/432f32f8097b03e3cd5fe57e88efb685d964e2e5178a48ed61e841f7fdce/pyyaml_env_tag-1.1-py3-none-any.whl", hash = "sha256:17109e1a528561e32f026364712fee1264bc2ea6715120891174ed1b980d2e04", size = 4722, upload-time = "2025-05-13T15:23:59.629Z" },
 ]
 
+[[package]]
+name = "reactivex"
+version = "4.1.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/b6/af/38a4b62468e4c5bd50acf511d86fe62e65a466aa6abb55b1d59a4a9e57f3/reactivex-4.1.0.tar.gz", hash = "sha256:c7499e3c802bccaa20839b3e17355a7d939573fded3f38ba3d4796278a169a3d", size = 113482, upload-time = "2025-11-05T21:44:24.557Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ba/9e/3c2f5d3abb6c5d82f7696e1e3c69b7279049e928596ce82ed25ca97a08f3/reactivex-4.1.0-py3-none-any.whl", hash = "sha256:485750ec8d9b34bcc8ff4318971d234dc4f595058a1b4435a74aefef4b2bc9bd", size = 218588, upload-time = "2025-11-05T21:44:23.015Z" },
+]
+
 [[package]]
 name = "requests"
 version = "2.34.2"