Event handling

README.md

@@ -19,6 +19,7 @@ We ship three binaries that mirror Lua scripts:
 - `examples/list-devices.rs`
 - `examples/invoke-method.rs`
 - `examples/redstone.rs`
+- `examples/redstone-events.rs`
 
 Build them with:
 

docs/index.md

@@ -37,8 +37,8 @@ fn main() -> Result<()> {
 
     // Subscribe to events and drain one.
     device.subscribe()?;
-    let DeviceEvent { device_id, payload } = bus.next_event()?;
-    println!("{device_id} -> {payload}");
+    let event = bus.next_event_for(&device_id)?;
+    println!("{device_id} -> {:#?}", event.payload);
 
     Ok(())
 }
@@ -52,9 +52,9 @@ fn main() -> Result<()> {
   implements `IntoJsonArgs` (tuples, slices, `Vec<T>`) is accepted and converted
   into JSON automatically.
 - **Subscriptions:** `Device::subscribe` / `Device::unsubscribe` send the
-  corresponding RPC requests. `DeviceBus::next_event` blocks until a queued event
-  is available; `DeviceBus::try_event` checks the queue without blocking. Events
-  surface as `DeviceEvent { device_id, payload }`.
+  corresponding RPC requests. `DeviceBus::next_event_for(device_id)` waits for
+  the next matching event; `Device::next_event()` provides a convenience wrapper.
+  `try_event*` variants check queues without blocking.
 
 ## Typed Wrappers
 
@@ -77,6 +77,7 @@ examples and method-by-method coverage.
 ## Additional Topics
 
 - [Device wrapper reference](wrappers.md)
+- [Redstone event example](../examples/redstone-events.rs)
 - [Performance comparison workflow](performance_testing.md)
 
 Contributions and further documentation improvements are welcome. Open an issue

docs/wrappers.md

@@ -13,20 +13,28 @@ Before using any wrapper, attach it:
 let mut bus = DeviceBus::connect(DEFAULT_DEVICE_PATH)?;
 let mut redstone = oc2r_rust::Redstone::attach(&mut bus)?
     .expect("no redstone interface attached");
+// call unsubscribe() once you are done processing events
+redstone.subscribe()?;
 ```
 
 ## Redstone
 
 ```rust
-use oc2r_rust::{Redstone, Side};
+use oc2r_rust::{Redstone, RedstoneSignal, Side};
 
 redstone.set_output_state(Side::East, true)?;
 let input = redstone.input(Side::East)?;
 let output = redstone.output(Side::East)?;
 redstone.set_output_level(Side::East, 7)?;
+
+if let Some(signal) = redstone.try_event()? {
+    println!("{} -> {}", signal.side, signal.level);
+}
 ```
 
 `Side` implements `FromStr` so strings like "north" can be parsed safely.
+Incoming events surface as `RedstoneSignal { side, level }`; see also the
+`examples/redstone-events.rs` program.
 
 ## Energy Storage
 
@@ -135,5 +143,9 @@ if fie.request_import()? {
 }
 ```
 
+The raw event payloads for these devices can also be obtained via
+`Device::next_event`/`DeviceBus::next_event_for` when you need to handle
+custom packets.
+
 Feel free to extend these wrappers or add new ones; each uses the same
 `define_wrapper!` macro found in `src/devices/mod.rs`.

examples/redstone-events.rs

@@ -0,0 +1,18 @@
+use oc2r_rust::{DEFAULT_DEVICE_PATH, DeviceBus, Redstone, RedstoneSignal, Result};
+
+fn main() -> Result<()> {
+    let mut bus = DeviceBus::connect(DEFAULT_DEVICE_PATH)?;
+    let mut redstone = Redstone::attach(&mut bus)?.expect("no redstone interface");
+    redstone.subscribe()?;
+
+    println!("Waiting for redstone events. Toggle any side to exit.");
+    loop {
+        let signal: RedstoneSignal = redstone.next_event()?;
+        println!("{} -> {}", signal.side, signal.level);
+        if signal.level == 0 {
+            break;
+        }
+    }
+
+    Ok(())
+}

src/bus.rs

@@ -160,20 +160,65 @@ impl DeviceBus {
         self.subscribe_request(RpcRequest::Unsubscribe { data: device_id })
     }
 
-    pub fn next_event(&mut self) -> Result<DeviceEvent> {
-        if let Some(event) = self.events.pop_front() {
-            return Ok(event);
-        }
-        match read_message(self.fd())? {
-            RpcResponse::Event { data } => DeviceEvent::from_json(data),
-            other => Err(Error::Protocol(format!(
-                "unexpected response while waiting for event: {other:?}"
-            ))),
+    pub fn next_event_for(&mut self, device_id: &str) -> Result<DeviceEvent> {
+        self.next_event_where(|event| event.device_id == device_id)
+    }
+
+    pub fn try_event_for(&mut self, device_id: &str) -> Option<DeviceEvent> {
+        self.try_event_where(|event| event.device_id == device_id)
+    }
+
+    fn pop_matching_event<F>(&mut self, predicate: &mut F) -> Option<DeviceEvent>
+    where
+        F: FnMut(&DeviceEvent) -> bool,
+    {
+        if let Some(index) = self.events.iter().position(predicate) {
+            self.events.remove(index)
+        } else {
+            None
         }
     }
 
+    fn next_event_where<F>(&mut self, mut predicate: F) -> Result<DeviceEvent>
+    where
+        F: FnMut(&DeviceEvent) -> bool,
+    {
+        if let Some(event) = self.pop_matching_event(&mut predicate) {
+            return Ok(event);
+        }
+
+        loop {
+            match read_message(self.fd())? {
+                RpcResponse::Event { data } => {
+                    let event = DeviceEvent::from_json(data)?;
+                    if predicate(&event) {
+                        return Ok(event);
+                    } else {
+                        self.events.push_back(event);
+                    }
+                }
+                other => {
+                    return Err(Error::Protocol(format!(
+                        "unexpected response while waiting for event: {other:?}"
+                    )));
+                }
+            }
+        }
+    }
+
+    fn try_event_where<F>(&mut self, mut predicate: F) -> Option<DeviceEvent>
+    where
+        F: FnMut(&DeviceEvent) -> bool,
+    {
+        self.pop_matching_event(&mut predicate)
+    }
+
+    pub fn next_event(&mut self) -> Result<DeviceEvent> {
+        self.next_event_where(|_| true)
+    }
+
     pub fn try_event(&mut self) -> Option<DeviceEvent> {
-        self.events.pop_front()
+        self.try_event_where(|_| true)
     }
 
     pub fn get(&mut self, device_id: &str) -> Result<Option<DeviceEntry>> {
@@ -253,4 +298,12 @@ impl<'bus> Device<'bus> {
     pub fn unsubscribe(&mut self) -> Result<()> {
         self.bus.unsubscribe(&self.entry.device_id)
     }
+
+    pub fn next_event(&mut self) -> Result<DeviceEvent> {
+        self.bus.next_event_for(&self.entry.device_id)
+    }
+
+    pub fn try_event(&mut self) -> Option<DeviceEvent> {
+        self.bus.try_event_for(&self.entry.device_id)
+    }
 }

src/devices/mod.rs

@@ -61,5 +61,5 @@ pub use file_import_export::{FileImportExport, ImportedFileInfo};
 pub use fluid_handler::FluidHandler;
 pub use inventory_operations::InventoryOperations;
 pub use item_handler::ItemHandler;
-pub use redstone::{ParseSideError, Redstone, Side};
+pub use redstone::{ParseSideError, Redstone, RedstoneSignal, Side};
 pub use sound::SoundCard;

src/devices/redstone.rs

@@ -1,5 +1,5 @@
-use crate::Result;
 use crate::value::JsonValueExt;
+use crate::{DeviceEvent, Error, Result};
 use std::fmt;
 use std::str::FromStr;
 
@@ -36,6 +36,26 @@ impl<'bus> Redstone<'bus> {
     pub fn set_output_state(&mut self, side: Side, active: bool) -> Result<()> {
         self.set_output_level(side, if active { 15 } else { 0 })
     }
+
+    pub fn subscribe(&mut self) -> Result<()> {
+        self.device.subscribe()
+    }
+
+    pub fn unsubscribe(&mut self) -> Result<()> {
+        self.device.unsubscribe()
+    }
+
+    pub fn next_event(&mut self) -> Result<RedstoneSignal> {
+        let event = self.device.next_event()?;
+        RedstoneSignal::from_device_event(event)
+    }
+
+    pub fn try_event(&mut self) -> Result<Option<RedstoneSignal>> {
+        match self.device.try_event() {
+            Some(event) => RedstoneSignal::from_device_event(event).map(Some),
+            None => Ok(None),
+        }
+    }
 }
 
 /// Redstone interface sides used by OC2R.
@@ -107,3 +127,41 @@ impl fmt::Display for ParseSideError {
 }
 
 impl std::error::Error for ParseSideError {}
+
+#[derive(Debug, Clone, PartialEq)]
+pub struct RedstoneSignal {
+    pub side: Side,
+    pub level: i32,
+}
+
+impl RedstoneSignal {
+    fn from_device_event(event: DeviceEvent) -> Result<Self> {
+        let mut object = event.payload.into_object("redstone event")?;
+        let event_name = object
+            .remove("event")
+            .ok_or_else(|| Error::Protocol("missing event field".into()))?
+            .into_string("redstone event name")?;
+        if event_name != "redstone" {
+            return Err(Error::Protocol(format!(
+                "unexpected redstone event type {event_name}"
+            )));
+        }
+
+        let side_str = object
+            .remove("side")
+            .ok_or_else(|| Error::Protocol("missing side field".into()))?
+            .into_string("redstone event side")?;
+        let side = Side::from_str(&side_str)
+            .map_err(|_| Error::Protocol(format!("invalid redstone event side {side_str}")))?;
+
+        let level = object
+            .remove("level")
+            .ok_or_else(|| Error::Protocol("missing level field".into()))?
+            .into_i64("redstone event level")?;
+
+        Ok(Self {
+            side,
+            level: level as i32,
+        })
+    }
+}

src/lib.rs

@@ -34,7 +34,8 @@ mod value;
 pub use bus::{Device, DeviceBus, DeviceEvent};
 pub use devices::{
     BlockOperations, Cpu, EnergyStorage, FileImportExport, FluidHandler, ImportedFileInfo,
-    InventoryOperations, ItemHandler, ParseSideError, Redstone, RobotSide, Side, SoundCard,
+    InventoryOperations, ItemHandler, ParseSideError, Redstone, RedstoneSignal, RobotSide, Side,
+    SoundCard,
 };
 pub use error::{Error, Result};
 pub use rpc::{DeviceEntry, JsonValue, MethodEntry, MethodParameter};