Struct LiveDataStream

pub struct LiveDataStream<R: Read + Unpin, W: Write + Unpin> { /* private fields */ }

A Stream/Sink wrapper for RESOL VBus Data items encoded in the live / wire representation.

It also contains methods to communicate with a VBus device to get or set values etc.

Implementations

impl<R: Read + Unpin, W: Write + Unpin> LiveDataStream<R, W>

pub fn new( reader: R, writer: W, channel: u8, self_address: u16, ) -> LiveDataStream<R, W>

Create a new LiveDataStream.

Examples found in repository

examples/live_data_recorder.rs (line 26)

fn main() -> Result<()> {
    async_std::task::block_on(async {
        // Create an recording file and hand it to a `RecordingWriter`
        let file = File::create("test.vbus")?;
        let mut rw = RecordingWriter::new(file);

        // Parse the address of the DL2 to connect to
        let addr = "192.168.5.217:7053".parse::<SocketAddr>()?;

        let stream = TcpStream::connect(addr).await?;

        let mut hs = TcpClientHandshake::start(stream).await?;
        hs.send_pass_command("vbus").await?;
        let stream = hs.send_data_command().await?;

        let (reader, writer) = (&stream, &stream);

        let mut stream = LiveDataStream::new(reader, writer, 0, 0x0020);

        while let Some(data) = stream.receive_any_data(60000).await? {
            println!("{}", data.id_string());

            // Add `Data` value into `DataSet` to be stored
            let mut data_set = DataSet::new();
            data_set.timestamp = data.as_ref().timestamp;
            data_set.add_data(data);

            // Write the `DataSet` into the `RecordingWriter` for permanent storage
            rw.write_data_set(&data_set)
                .expect("Unable to write data set");
        }

        Ok(())
    })
}

pub fn into_inner(self) -> (R, W)

Consume self and return the underlying I/O pair.

pub async fn receive<F>( &mut self, timeout_ms: u64, filter: F, ) -> Result<Option<Data>>

where F: Fn(&Data) -> bool,

Receive data from the VBus.

This methods waits for timeout_ms milliseconds for incoming VBus data. Every time a valid Data is received over the VBus the filter function is called with that Data as its argument. The function returns a bool whether the provided Data is the data it was waiting for.

If the filter function returns true, the respective Data is used to return from the receive method.

If the filter function did not find the matching data within timeout_ms milliseconds, the receive method returns with None.

pub async fn transceive<F>( &mut self, tx_data: Data, max_tries: usize, initial_timeout_ms: u64, timeout_increment_ms: u64, filter: F, ) -> Result<Option<Data>>

where F: Fn(&Data) -> bool,

Send data to the VBus and wait for a reply.

This method sends the tx_data to the VBus and waits for up to initial_timeout_ms milliseconds for a reply.

Every time a valid Data is received over the VBus the filter function is called with that Data as its argument. The function returns a bool whether the provided Data is the reply it was waiting for.

If the filter function returns true, the respective Data is used to return from the transceive method.

If the filter function did not find the matching reply within initial_timeout_ms milliseconds, the tx_data is send again up max_tries times, increasing the timeout by timeout_increment_ms milliseconds every time.

After max_tries without a matching reply the transceive method returns with None.

pub async fn receive_any_data( &mut self, timeout_ms: u64, ) -> Result<Option<Data>>

Wait for any VBus data.

Examples found in repository

examples/live_data_recorder.rs (line 28)

fn main() -> Result<()> {
    async_std::task::block_on(async {
        // Create an recording file and hand it to a `RecordingWriter`
        let file = File::create("test.vbus")?;
        let mut rw = RecordingWriter::new(file);

        // Parse the address of the DL2 to connect to
        let addr = "192.168.5.217:7053".parse::<SocketAddr>()?;

        let stream = TcpStream::connect(addr).await?;

        let mut hs = TcpClientHandshake::start(stream).await?;
        hs.send_pass_command("vbus").await?;
        let stream = hs.send_data_command().await?;

        let (reader, writer) = (&stream, &stream);

        let mut stream = LiveDataStream::new(reader, writer, 0, 0x0020);

        while let Some(data) = stream.receive_any_data(60000).await? {
            println!("{}", data.id_string());

            // Add `Data` value into `DataSet` to be stored
            let mut data_set = DataSet::new();
            data_set.timestamp = data.as_ref().timestamp;
            data_set.add_data(data);

            // Write the `DataSet` into the `RecordingWriter` for permanent storage
            rw.write_data_set(&data_set)
                .expect("Unable to write data set");
        }

        Ok(())
    })
}

pub async fn wait_for_free_bus(&mut self) -> Result<Option<Datagram>>

Wait for a datagram that offers VBus control.

pub async fn release_bus(&mut self, address: u16) -> Result<Option<Data>>

Give back bus control to the regular VBus master.

pub async fn get_value_by_index( &mut self, address: u16, index: i16, subindex: u8, ) -> Result<Option<Datagram>>

Get a value by its index.

pub async fn set_value_by_index( &mut self, address: u16, index: i16, subindex: u8, value: i32, ) -> Result<Option<Datagram>>

Set a value by its index.

pub async fn get_value_id_hash_by_index( &mut self, address: u16, index: i16, ) -> Result<Option<Datagram>>

Get a value’s ID hash by its index.

pub async fn get_value_index_by_id_hash( &mut self, address: u16, id_hash: i32, ) -> Result<Option<Datagram>>

Get a value’s index by its ID hash.

pub async fn get_caps1(&mut self, address: u16) -> Result<Option<Datagram>>

Get the capabilities (part 1) from a VBus device.

pub async fn begin_bulk_value_transaction( &mut self, address: u16, tx_timeout: i32, ) -> Result<Option<Datagram>>

Begin a bulk value transaction.

pub async fn commit_bulk_value_transaction( &mut self, address: u16, ) -> Result<Option<Datagram>>

Commit a bulk value transaction.

pub async fn rollback_bulk_value_transaction( &mut self, address: u16, ) -> Result<Option<Datagram>>

Rollback a bulk value transaction.

pub async fn set_bulk_value_by_index( &mut self, address: u16, index: i16, subindex: u8, value: i32, ) -> Result<Option<Datagram>>

Set a value by its index while inside a bulk value transaction.

Trait Implementations

impl<R: Debug + Read + Unpin, W: Debug + Write + Unpin> Debug for LiveDataStream<R, W>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.