rodbus 1.4.0

A high-performance async implementation of the Modbus protocol using tokio
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

use crate::common::function::FunctionCode;
use crate::common::traits::Serialize;
use crate::decode::AppDecodeLevel;
use crate::error::RequestError;
use crate::types::{
    AddressRange, Indexed, ReadRegistersRange, RegisterIterator, RegisterIteratorDisplay,
};

use scursor::{ReadCursor, WriteCursor};

pub(crate) trait RegistersCallback:
    FnOnce(Result<RegisterIterator, RequestError>) + Send + Sync + 'static
{
}
impl<T> RegistersCallback for T where
    T: FnOnce(Result<RegisterIterator, RequestError>) + Send + Sync + 'static
{
}

pub(crate) struct Promise {
    callback: Option<Box<dyn RegistersCallback>>,
}

impl Drop for Promise {
    fn drop(&mut self) {
        self.failure(RequestError::Shutdown);
    }
}

impl Promise {
    pub(crate) fn new<T>(callback: T) -> Self
    where
        T: RegistersCallback,
    {
        Self {
            callback: Some(Box::new(callback)),
        }
    }

    pub(crate) fn failure(&mut self, err: RequestError) {
        self.complete(Err(err))
    }

    pub(crate) fn success(&mut self, iter: RegisterIterator) {
        self.complete(Ok(iter))
    }

    fn complete(&mut self, x: Result<RegisterIterator, RequestError>) {
        if let Some(callback) = self.callback.take() {
            callback(x)
        }
    }
}

pub(crate) struct ReadRegisters {
    pub(crate) request: ReadRegistersRange,
    promise: Promise,
}

impl ReadRegisters {
    pub(crate) fn new(request: ReadRegistersRange, promise: Promise) -> Self {
        Self { request, promise }
    }

    pub(crate) fn channel(
        request: ReadRegistersRange,
        tx: tokio::sync::oneshot::Sender<Result<Vec<Indexed<u16>>, RequestError>>,
    ) -> Self {
        Self::new(
            request,
            Promise::new(|x: Result<RegisterIterator, RequestError>| {
                let _ = tx.send(x.map(|x| x.collect()));
            }),
        )
    }

    pub(crate) fn serialize(&self, cursor: &mut WriteCursor) -> Result<(), RequestError> {
        self.request.get().serialize(cursor)
    }

    pub(crate) fn failure(&mut self, err: RequestError) {
        self.promise.failure(err)
    }

    pub(crate) fn handle_response(
        &mut self,
        mut cursor: ReadCursor,
        function: FunctionCode,
        decode: AppDecodeLevel,
    ) -> Result<(), RequestError> {
        let response = Self::parse_registers_response(self.request.get(), &mut cursor)?;

        if decode.enabled() {
            tracing::info!(
                "PDU RX - {} {}",
                function,
                RegisterIteratorDisplay::new(decode, response)
            );
        }

        self.promise.success(response);
        Ok(())
    }

    fn parse_registers_response<'a>(
        range: AddressRange,
        cursor: &'a mut ReadCursor,
    ) -> Result<RegisterIterator<'a>, RequestError> {
        // there's a byte-count here that we don't actually need
        cursor.read_u8()?;
        // the reset is a sequence of bits
        RegisterIterator::parse_all(range, cursor)
    }
}