pamoja_modbus/
client.rs

1//! A Modbus client on a serial line: the gateway's side of the conversation.
2
3use std::fmt;
4use std::time::{Duration, Instant};
5use std::vec::Vec;
6
7use pamoja_hal::port::{PortError, PortKind, SerialPort, Settings};
8
9use crate::adu::Adu;
10use crate::error::ModbusError;
11use crate::function::Exception;
12use crate::pdu::Pdu;
13use crate::request::Request;
14use crate::server::BROADCAST;
15
16/// How long a client waits for a reply unless told otherwise: one second, the low end of what
17/// the serial line specification suggests at 9600 baud.
18pub const RESPONSE_TIMEOUT: Duration = Duration::from_secs(1);
19
20/// How long a client leaves the line quiet after a broadcast unless told otherwise, so every
21/// device has carried it out before the next request: 100 milliseconds, the low end of the
22/// range the serial line specification gives.
23pub const TURNAROUND: Duration = Duration::from_millis(100);
24
25/// Why a transaction failed.
26#[derive(Clone, Debug, PartialEq, Eq)]
27pub enum ClientError {
28    /// The request could not be built: a quantity outside what one request carries.
29    Request(ModbusError),
30    /// A read was addressed to the broadcast address, which no device answers.
31    BroadcastRead,
32    /// The serial port failed.
33    Port(PortError),
34    /// No complete reply arrived within the response timeout.
35    Timeout {
36        /// The unit that was asked.
37        unit: u8,
38        /// How many bytes of a reply had arrived.
39        received: usize,
40    },
41    /// The reply failed its CRC or is not the shape its function gives.
42    Frame(ModbusError),
43    /// A reply came back from a unit other than the one asked.
44    WrongUnit {
45        /// The unit that was asked.
46        expected: u8,
47        /// The unit that answered.
48        found: u8,
49    },
50    /// A reply answered a function other than the one asked.
51    WrongFunction {
52        /// The function that was asked.
53        expected: u8,
54        /// The function the reply names.
55        found: u8,
56    },
57    /// A well-formed reply that does not answer the request: a write echoed other values, or
58    /// a read returned another count.
59    Mismatch {
60        /// The unit that answered.
61        unit: u8,
62    },
63    /// The device refused the request.
64    Exception {
65        /// The unit that refused.
66        unit: u8,
67        /// The function it refused.
68        function: u8,
69        /// Why.
70        exception: Exception,
71    },
72}
73
74impl fmt::Display for ClientError {
75    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
76        match self {
77            ClientError::Request(error) => write!(f, "the request cannot be sent: {error}"),
78            ClientError::BroadcastRead => {
79                f.write_str("a read cannot be broadcast, since no device answers a broadcast")
80            }
81            ClientError::Port(error) => write!(f, "the serial port failed: {error}"),
82            ClientError::Timeout { unit, received: 0 } => {
83                write!(f, "unit {unit} did not answer within the response timeout")
84            }
85            ClientError::Timeout { unit, received } => write!(
86                f,
87                "unit {unit} sent {received} bytes of a reply and then went quiet"
88            ),
89            ClientError::Frame(error) => write!(f, "the reply is not a valid frame: {error}"),
90            ClientError::WrongUnit { expected, found } => {
91                write!(f, "unit {found} answered a request for unit {expected}")
92            }
93            ClientError::WrongFunction { expected, found } => write!(
94                f,
95                "the reply names function {found:#04x}, not the {expected:#04x} asked for"
96            ),
97            ClientError::Mismatch { unit } => {
98                write!(
99                    f,
100                    "unit {unit} answered with values the request did not ask for"
101                )
102            }
103            ClientError::Exception {
104                unit,
105                function,
106                exception,
107            } => write!(
108                f,
109                "unit {unit} refused function {function:#04x} with exception {:#04x}, {exception}",
110                exception.code()
111            ),
112        }
113    }
114}
115
116impl std::error::Error for ClientError {
117    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
118        match self {
119            ClientError::Request(error) | ClientError::Frame(error) => Some(error),
120            ClientError::Port(error) => Some(error),
121            _ => None,
122        }
123    }
124}
125
126/// A Modbus RTU client on a serial line: the gateway, the master in the specification's
127/// words, that sends each request and waits for its reply.
128///
129/// Each transaction follows the Modbus over Serial Line specification. The client leaves
130/// the line silent for 3.5 characters, or 1.75 ms above 19200 baud, so the request starts a
131/// new frame; drops anything stale waiting in the port; writes the request; and reads the
132/// reply to the length the request implies, against the response timeout. The reply's CRC,
133/// unit, and function are checked before a value is read out of it, and a refusal comes
134/// back as [`ClientError::Exception`]. A write to [`BROADCAST`] reaches every device and
135/// draws no reply, so the client waits out the turnaround delay instead. The specification
136/// reserves units 248 to 255, and a client still sends to them, since some devices answer on
137/// one.
138///
139/// On the kernel's serial device every wait is real. On a simulated, paired, or scripted
140/// port nothing waits: the waits are counted in the port's
141/// [`waited_micros`](SerialPort::waited_micros), so a test of a meter that never answers
142/// runs at once and still says how long a real one would have taken.
143///
144/// # Examples
145///
146/// ```
147/// use std::sync::{Arc, Mutex};
148///
149/// use pamoja_hal::port::{Parity, SerialPort, Settings};
150/// use pamoja_modbus::{Client, Server};
151///
152/// // A power meter at unit 17, simulated on the far end of a 19200 8E1 line.
153/// let meter = Server::new(17)?.with_holding_registers(107, &[2301, 418, 0]);
154/// let meter = Arc::new(Mutex::new(meter));
155/// let line = Settings::new(19_200).with_parity(Parity::Even);
156/// let mut client = Client::new(SerialPort::simulated(line, Arc::clone(&meter)));
157///
158/// assert_eq!(client.read_holding_registers(17, 107, 3)?, [2301, 418, 0]);
159/// client.write_single_register(17, 109, 1)?;
160/// assert_eq!(meter.lock().unwrap().holding_register(109), Some(1));
161/// # Ok::<(), Box<dyn std::error::Error>>(())
162/// ```
163#[derive(Debug)]
164pub struct Client {
165    port: SerialPort,
166    response_timeout: Duration,
167    turnaround: Duration,
168}
169
170enum Expect {
171    Bits { quantity: u16 },
172    Registers { quantity: u16 },
173    Echo,
174}
175
176impl Client {
177    /// Makes a client on a port, with a one-second response timeout and a 100 ms turnaround.
178    ///
179    /// # Arguments
180    ///
181    /// * `port` - the line, opened at the speed and format the devices on it use.
182    ///
183    /// # Returns
184    ///
185    /// The client.
186    pub fn new(port: SerialPort) -> Client {
187        Client {
188            port,
189            response_timeout: RESPONSE_TIMEOUT,
190            turnaround: TURNAROUND,
191        }
192    }
193
194    /// Sets how long the client waits for a whole reply once a request has gone out.
195    ///
196    /// # Arguments
197    ///
198    /// * `timeout` - the response timeout; a slow device or a slow line needs more.
199    ///
200    /// # Returns
201    ///
202    /// The client.
203    #[must_use]
204    pub fn with_response_timeout(mut self, timeout: Duration) -> Client {
205        self.response_timeout = timeout;
206        self
207    }
208
209    /// Sets how long the client leaves the line quiet after a broadcast.
210    ///
211    /// # Arguments
212    ///
213    /// * `delay` - the turnaround delay, long enough for the slowest device to carry out
214    ///   the write.
215    ///
216    /// # Returns
217    ///
218    /// The client.
219    #[must_use]
220    pub fn with_turnaround(mut self, delay: Duration) -> Client {
221        self.turnaround = delay;
222        self
223    }
224
225    /// Changes how long the client waits for a whole reply, as
226    /// [`with_response_timeout`](Client::with_response_timeout) does when the client is made.
227    ///
228    /// # Arguments
229    ///
230    /// * `timeout` - the response timeout.
231    pub fn set_response_timeout(&mut self, timeout: Duration) {
232        self.response_timeout = timeout;
233    }
234
235    /// Changes how long the client leaves the line quiet after a broadcast, as
236    /// [`with_turnaround`](Client::with_turnaround) does when the client is made.
237    ///
238    /// # Arguments
239    ///
240    /// * `delay` - the turnaround delay.
241    pub fn set_turnaround(&mut self, delay: Duration) {
242        self.turnaround = delay;
243    }
244
245    /// Returns the port the client runs on.
246    pub fn port(&self) -> &SerialPort {
247        &self.port
248    }
249
250    /// Returns how long the client waits for a reply.
251    pub fn response_timeout(&self) -> Duration {
252        self.response_timeout
253    }
254
255    /// Returns how long the client leaves the line quiet after a broadcast.
256    pub fn turnaround(&self) -> Duration {
257        self.turnaround
258    }
259
260    /// Returns the silence that separates two frames, 3.5 characters at the line's speed and
261    /// format, and a fixed 1.75 ms above 19200 baud, where the specification stops scaling it.
262    ///
263    /// # Arguments
264    ///
265    /// * `settings` - the line's speed and character format.
266    ///
267    /// # Returns
268    ///
269    /// The silence, rounded up to the nanosecond.
270    pub fn frame_gap(settings: Settings) -> Duration {
271        if settings.baud > 19_200 {
272            Duration::from_micros(1_750)
273        } else {
274            Duration::from_nanos((settings.character_nanos() * 7).div_ceil(2))
275        }
276    }
277
278    /// Reads coils, function `0x01`.
279    ///
280    /// # Arguments
281    ///
282    /// * `unit` - the device, 1 to 247.
283    /// * `start` - the first coil's address.
284    /// * `quantity` - how many, 1 to 2000.
285    ///
286    /// # Returns
287    ///
288    /// The coils' states, in address order.
289    ///
290    /// # Errors
291    ///
292    /// Any [`ClientError`]: a quantity out of range, a broadcast, a port failure, a timeout,
293    /// a reply that fails its checks, or the device's exception.
294    pub fn read_coils(
295        &mut self,
296        unit: u8,
297        start: u16,
298        quantity: u16,
299    ) -> Result<Vec<bool>, ClientError> {
300        self.read_bits(unit, Pdu::read_coils(start, quantity), quantity)
301    }
302
303    /// Reads discrete inputs, function `0x02`.
304    ///
305    /// # Arguments
306    ///
307    /// * `unit` - the device, 1 to 247.
308    /// * `start` - the first input's address.
309    /// * `quantity` - how many, 1 to 2000.
310    ///
311    /// # Returns
312    ///
313    /// The inputs' states, in address order.
314    ///
315    /// # Errors
316    ///
317    /// Any [`ClientError`], as for [`read_coils`](Client::read_coils).
318    pub fn read_discrete_inputs(
319        &mut self,
320        unit: u8,
321        start: u16,
322        quantity: u16,
323    ) -> Result<Vec<bool>, ClientError> {
324        self.read_bits(unit, Pdu::read_discrete_inputs(start, quantity), quantity)
325    }
326
327    /// Reads holding registers, function `0x03`.
328    ///
329    /// # Arguments
330    ///
331    /// * `unit` - the device, 1 to 247.
332    /// * `start` - the first register's address.
333    /// * `quantity` - how many, 1 to 125.
334    ///
335    /// # Returns
336    ///
337    /// The registers' values, in address order.
338    ///
339    /// # Errors
340    ///
341    /// Any [`ClientError`], as for [`read_coils`](Client::read_coils).
342    pub fn read_holding_registers(
343        &mut self,
344        unit: u8,
345        start: u16,
346        quantity: u16,
347    ) -> Result<Vec<u16>, ClientError> {
348        let pdu = Pdu::read_holding_registers(start, quantity);
349        self.read_words(unit, pdu, quantity)
350    }
351
352    /// Reads input registers, function `0x04`.
353    ///
354    /// # Arguments
355    ///
356    /// * `unit` - the device, 1 to 247.
357    /// * `start` - the first register's address.
358    /// * `quantity` - how many, 1 to 125.
359    ///
360    /// # Returns
361    ///
362    /// The registers' values, in address order.
363    ///
364    /// # Errors
365    ///
366    /// Any [`ClientError`], as for [`read_coils`](Client::read_coils).
367    pub fn read_input_registers(
368        &mut self,
369        unit: u8,
370        start: u16,
371        quantity: u16,
372    ) -> Result<Vec<u16>, ClientError> {
373        let pdu = Pdu::read_input_registers(start, quantity);
374        self.read_words(unit, pdu, quantity)
375    }
376
377    /// Writes one coil, function `0x05`.
378    ///
379    /// # Arguments
380    ///
381    /// * `unit` - the device, 1 to 247, or [`BROADCAST`] for every device.
382    /// * `address` - the coil's address.
383    /// * `on` - the state to write.
384    ///
385    /// # Errors
386    ///
387    /// Any [`ClientError`] but [`BroadcastRead`](ClientError::BroadcastRead).
388    pub fn write_single_coil(
389        &mut self,
390        unit: u8,
391        address: u16,
392        on: bool,
393    ) -> Result<(), ClientError> {
394        let pdu = Pdu::write_single_coil(address, on);
395        self.write(unit, pdu, pdu)
396    }
397
398    /// Writes one holding register, function `0x06`.
399    ///
400    /// # Arguments
401    ///
402    /// * `unit` - the device, 1 to 247, or [`BROADCAST`] for every device.
403    /// * `address` - the register's address.
404    /// * `value` - the value to write.
405    ///
406    /// # Errors
407    ///
408    /// Any [`ClientError`] but [`BroadcastRead`](ClientError::BroadcastRead).
409    pub fn write_single_register(
410        &mut self,
411        unit: u8,
412        address: u16,
413        value: u16,
414    ) -> Result<(), ClientError> {
415        let pdu = Pdu::write_single_register(address, value);
416        self.write(unit, pdu, pdu)
417    }
418
419    /// Writes a run of coils, function `0x0F`.
420    ///
421    /// # Arguments
422    ///
423    /// * `unit` - the device, 1 to 247, or [`BROADCAST`] for every device.
424    /// * `start` - the first coil's address.
425    /// * `values` - the states to write, 1 to 1968 of them, in address order.
426    ///
427    /// # Errors
428    ///
429    /// Any [`ClientError`] but [`BroadcastRead`](ClientError::BroadcastRead).
430    pub fn write_multiple_coils(
431        &mut self,
432        unit: u8,
433        start: u16,
434        values: &[bool],
435    ) -> Result<(), ClientError> {
436        let pdu = Pdu::write_multiple_coils(start, values).map_err(ClientError::Request)?;
437        let echo = Pdu::write_multiple_coils_reply(start, values.len() as u16);
438        self.write(unit, pdu, echo)
439    }
440
441    /// Writes a run of holding registers, function `0x10`.
442    ///
443    /// # Arguments
444    ///
445    /// * `unit` - the device, 1 to 247, or [`BROADCAST`] for every device.
446    /// * `start` - the first register's address.
447    /// * `values` - the values to write, 1 to 123 of them, in address order.
448    ///
449    /// # Errors
450    ///
451    /// Any [`ClientError`] but [`BroadcastRead`](ClientError::BroadcastRead).
452    pub fn write_multiple_registers(
453        &mut self,
454        unit: u8,
455        start: u16,
456        values: &[u16],
457    ) -> Result<(), ClientError> {
458        let pdu = Pdu::write_multiple_registers(start, values).map_err(ClientError::Request)?;
459        let echo = Pdu::write_multiple_registers_reply(start, values.len() as u16);
460        self.write(unit, pdu, echo)
461    }
462
463    fn read_bits(&mut self, unit: u8, pdu: Pdu, quantity: u16) -> Result<Vec<bool>, ClientError> {
464        let reply = self
465            .transact(unit, pdu, Expect::Bits { quantity })?
466            .ok_or(ClientError::BroadcastRead)?;
467        let bits = reply
468            .response()
469            .coils(quantity)
470            .map_err(|_| ClientError::Mismatch { unit })?;
471        Ok(bits.collect())
472    }
473
474    fn read_words(&mut self, unit: u8, pdu: Pdu, quantity: u16) -> Result<Vec<u16>, ClientError> {
475        let reply = self
476            .transact(unit, pdu, Expect::Registers { quantity })?
477            .ok_or(ClientError::BroadcastRead)?;
478        let words = reply
479            .response()
480            .registers()
481            .map_err(|_| ClientError::Frame(ModbusError::MalformedResponse))?;
482        if words.len() != usize::from(quantity) {
483            return Err(ClientError::Mismatch { unit });
484        }
485        Ok(words.collect())
486    }
487
488    fn write(&mut self, unit: u8, pdu: Pdu, echo: Pdu) -> Result<(), ClientError> {
489        match self.transact(unit, pdu, Expect::Echo)? {
490            Some(reply) if reply.pdu() != echo.as_bytes() => Err(ClientError::Mismatch { unit }),
491            _ => Ok(()),
492        }
493    }
494
495    // Runs one transaction, returning the reply, or nothing for a broadcast, which no device
496    // answers.
497    fn transact(&mut self, unit: u8, pdu: Pdu, expect: Expect) -> Result<Option<Adu>, ClientError> {
498        let request = Request::parse(pdu.as_bytes())
499            .map_err(|_| ClientError::Request(ModbusError::InvalidValueCount))?;
500        if unit == BROADCAST && request.reads() {
501            return Err(ClientError::BroadcastRead);
502        }
503        let frame = pdu.to_adu(unit);
504        self.port.wait(Client::frame_gap(self.port.settings()));
505        self.port.discard_input().map_err(ClientError::Port)?;
506        self.port
507            .write(frame.as_bytes())
508            .map_err(ClientError::Port)?;
509        if unit == BROADCAST {
510            self.port.wait(self.turnaround);
511            return Ok(None);
512        }
513        let reply = self.receive(unit, &expect)?;
514        let adu = Adu::parse(&reply).map_err(ClientError::Frame)?;
515        if adu.address() != unit {
516            return Err(ClientError::WrongUnit {
517                expected: unit,
518                found: adu.address(),
519            });
520        }
521        let function = pdu.function_code();
522        let found = adu.function_code();
523        if found & 0x7F != function {
524            return Err(ClientError::WrongFunction {
525                expected: function,
526                found,
527            });
528        }
529        if found & 0x80 != 0 {
530            let exception = adu
531                .exception()
532                .ok_or(ClientError::Frame(ModbusError::MalformedResponse))?;
533            return Err(ClientError::Exception {
534                unit,
535                function,
536                exception,
537            });
538        }
539        Ok(Some(adu))
540    }
541
542    // Reads one reply: the length a read's byte count gives, 8 bytes for a write's echo, and
543    // 5 for an exception, all against one response timeout.
544    fn receive(&mut self, unit: u8, expect: &Expect) -> Result<Vec<u8>, ClientError> {
545        let clock = self.port.kind() == PortKind::Device;
546        let started = Instant::now();
547        let mut frame = [0u8; Adu::MAX_LEN];
548        let mut want = match expect {
549            Expect::Bits { quantity, .. } => 5 + usize::from(*quantity).div_ceil(8),
550            Expect::Registers { quantity } => 5 + 2 * usize::from(*quantity),
551            Expect::Echo => 8,
552        };
553        let mut got = 0;
554        while got < want {
555            let timeout = if clock {
556                self.response_timeout.saturating_sub(started.elapsed())
557            } else {
558                self.response_timeout
559            };
560            let count = self
561                .port
562                .read(&mut frame[got..want], timeout)
563                .map_err(ClientError::Port)?;
564            if count == 0 {
565                return Err(ClientError::Timeout {
566                    unit,
567                    received: got,
568                });
569            }
570            got += count;
571            if got >= 2 && frame[1] & 0x80 != 0 {
572                want = 5;
573            } else if got >= 3 && !matches!(expect, Expect::Echo) {
574                want = (5 + usize::from(frame[2])).min(Adu::MAX_LEN);
575            }
576        }
577        Ok(frame[..want].to_vec())
578    }
579}
580
581#[cfg(test)]
582mod tests {
583    use super::*;
584    use std::sync::{Arc, Mutex};
585
586    use pamoja_hal::port::{Parity, PortStep, StopBits};
587
588    use crate::server::{Line, Server};
589
590    fn line() -> Settings {
591        Settings::new(19_200).with_parity(Parity::Even)
592    }
593
594    fn meter() -> Server {
595        Server::new(17)
596            .unwrap()
597            .with_holding_registers(107, &[2301, 418, 0])
598            .with_coils(0, &[false; 4])
599    }
600
601    #[test]
602    fn the_frame_gap_is_three_and_a_half_characters_up_to_19200_and_fixed_above() {
603        let at = |baud| Client::frame_gap(Settings::new(baud).with_parity(Parity::Even));
604        // 11 bits a character at 9600 is 1145.83 us, and 3.5 of them 4010.42 us.
605        assert_eq!(at(9_600), Duration::from_nanos(4_010_419));
606        assert_eq!(at(19_200), Duration::from_nanos(2_005_210));
607        for baud in [38_400, 57_600, 115_200] {
608            assert_eq!(at(baud), Duration::from_micros(1_750));
609        }
610        // No parity keeps 11 bits a character with a second stop bit.
611        let two = Settings::new(9_600).with_stop_bits(StopBits::Two);
612        assert_eq!(Client::frame_gap(two), at(9_600));
613    }
614
615    #[test]
616    fn a_read_puts_the_specification_frame_on_the_wire_and_decodes_the_reply() {
617        let reply = Pdu::read_holding_registers_reply(&[0x022B, 0x0000, 0x0064])
618            .unwrap()
619            .to_adu(0x11);
620        let port = SerialPort::scripted(
621            line(),
622            [
623                PortStep::write([0x11, 0x03, 0x00, 0x6B, 0x00, 0x03, 0x76, 0x87]),
624                PortStep::read(reply.as_bytes()),
625            ],
626        );
627        let mut client = Client::new(port.clone());
628        let values = client.read_holding_registers(0x11, 0x006B, 3).unwrap();
629        assert_eq!(values, [0x022B, 0x0000, 0x0064]);
630        assert_eq!(port.remaining(), Some(0));
631        assert_eq!(port.waited_micros(), 2_005, "only the frame gap was waited");
632    }
633
634    #[test]
635    fn every_function_round_trips_against_a_server() {
636        let meter = Arc::new(Mutex::new(
637            meter().with_discrete_inputs(0x00C4, &[true, false, true]),
638        ));
639        let mut client = Client::new(SerialPort::simulated(line(), Arc::clone(&meter)));
640        assert_eq!(
641            client.read_holding_registers(17, 107, 3).unwrap(),
642            [2301, 418, 0]
643        );
644        client.write_single_register(17, 109, 7).unwrap();
645        client
646            .write_multiple_registers(17, 107, &[2300, 420])
647            .unwrap();
648        assert_eq!(
649            client.read_holding_registers(17, 107, 3).unwrap(),
650            [2300, 420, 7]
651        );
652        client.write_single_coil(17, 1, true).unwrap();
653        client.write_multiple_coils(17, 2, &[true, true]).unwrap();
654        assert_eq!(
655            client.read_coils(17, 0, 4).unwrap(),
656            [false, true, true, true]
657        );
658        assert_eq!(
659            client.read_discrete_inputs(17, 0x00C4, 3).unwrap(),
660            [true, false, true]
661        );
662        let device = meter.lock().unwrap();
663        assert_eq!(device.holding_register(109), Some(7));
664        assert_eq!(device.served(), 8);
665    }
666
667    #[test]
668    fn a_refusal_comes_back_as_the_devices_exception() {
669        let mut client = Client::new(SerialPort::simulated(line(), meter()));
670        assert_eq!(
671            client.read_holding_registers(17, 108, 3),
672            Err(ClientError::Exception {
673                unit: 17,
674                function: 0x03,
675                exception: Exception::IllegalDataAddress
676            })
677        );
678        assert_eq!(
679            client.write_single_coil(17, 9, true),
680            Err(ClientError::Exception {
681                unit: 17,
682                function: 0x05,
683                exception: Exception::IllegalDataAddress
684            })
685        );
686    }
687
688    #[test]
689    fn a_unit_that_never_answers_times_out_after_the_response_timeout() {
690        let port = SerialPort::simulated(line(), meter());
691        let mut client =
692            Client::new(port.clone()).with_response_timeout(Duration::from_millis(250));
693        assert_eq!(
694            client.read_holding_registers(18, 107, 1),
695            Err(ClientError::Timeout {
696                unit: 18,
697                received: 0
698            })
699        );
700        assert_eq!(port.waited_micros(), 2_005 + 250_000);
701    }
702
703    #[test]
704    fn a_broadcast_write_waits_out_the_turnaround_and_every_device_carries_it_out() {
705        let mut line_of = Line::new();
706        let first = line_of.attach(meter());
707        let second = line_of.attach(Server::new(18).unwrap().with_holding_registers(109, &[0]));
708        let port = SerialPort::simulated(line(), line_of);
709        let mut client = Client::new(port.clone());
710        client.write_single_register(BROADCAST, 109, 1).unwrap();
711        client
712            .write_multiple_registers(BROADCAST, 109, &[1])
713            .unwrap();
714        assert_eq!(first.lock().unwrap().holding_register(109), Some(1));
715        assert_eq!(second.lock().unwrap().holding_register(109), Some(1));
716        assert_eq!(port.received(), 0, "nobody answers a broadcast");
717        assert_eq!(port.waited_micros(), 2 * (2_005 + 100_000));
718
719        assert_eq!(
720            client.read_holding_registers(BROADCAST, 109, 1),
721            Err(ClientError::BroadcastRead)
722        );
723        assert_eq!(port.written(), 8 + 11, "the broadcast read never went out");
724    }
725
726    #[test]
727    fn a_reply_that_fails_its_checks_is_refused_before_a_value_is_read() {
728        let request = Pdu::read_holding_registers(107, 1).to_adu(17);
729        let good = Pdu::read_holding_registers_reply(&[2301]).unwrap();
730
731        let mut corrupt = good.to_adu(17).as_bytes().to_vec();
732        corrupt[3] ^= 0x01;
733        let scripted = |reply: Vec<u8>| {
734            SerialPort::scripted(
735                line(),
736                [PortStep::write(request.as_bytes()), PortStep::read(reply)],
737            )
738        };
739        let mut client = Client::new(scripted(corrupt));
740        assert!(matches!(
741            client.read_holding_registers(17, 107, 1),
742            Err(ClientError::Frame(ModbusError::CrcMismatch { .. }))
743        ));
744
745        let mut client = Client::new(scripted(good.to_adu(18).as_bytes().to_vec()));
746        assert_eq!(
747            client.read_holding_registers(17, 107, 1),
748            Err(ClientError::WrongUnit {
749                expected: 17,
750                found: 18
751            })
752        );
753
754        let other = Pdu::read_input_registers_reply(&[2301]).unwrap().to_adu(17);
755        let mut client = Client::new(scripted(other.as_bytes().to_vec()));
756        assert_eq!(
757            client.read_holding_registers(17, 107, 1),
758            Err(ClientError::WrongFunction {
759                expected: 0x03,
760                found: 0x04
761            })
762        );
763
764        let two = Pdu::read_holding_registers_reply(&[2301, 418])
765            .unwrap()
766            .to_adu(17);
767        let mut client = Client::new(scripted(two.as_bytes().to_vec()));
768        assert_eq!(
769            client.read_holding_registers(17, 107, 1),
770            Err(ClientError::Mismatch { unit: 17 })
771        );
772
773        let half = good.to_adu(17).as_bytes()[..4].to_vec();
774        let mut client = Client::new(scripted(half));
775        assert_eq!(
776            client.read_holding_registers(17, 107, 1),
777            Err(ClientError::Timeout {
778                unit: 17,
779                received: 4
780            })
781        );
782    }
783
784    #[test]
785    fn a_request_the_protocol_cannot_carry_never_reaches_the_line() {
786        let port = SerialPort::simulated(line(), meter());
787        let mut client = Client::new(port.clone());
788        assert_eq!(
789            client.read_holding_registers(17, 0, 126),
790            Err(ClientError::Request(ModbusError::InvalidValueCount))
791        );
792        assert_eq!(
793            client.write_multiple_registers(17, 0, &[0; 124]),
794            Err(ClientError::Request(ModbusError::InvalidValueCount))
795        );
796        assert_eq!(port.written(), 0);
797    }
798
799    #[test]
800    fn a_reserved_unit_is_still_asked_since_some_devices_answer_on_one() {
801        let request = Pdu::read_input_registers(0, 1).to_adu(248);
802        let reply = Pdu::read_input_registers_reply(&[2301])
803            .unwrap()
804            .to_adu(248);
805        let port = SerialPort::scripted(
806            line(),
807            [
808                PortStep::write(request.as_bytes()),
809                PortStep::read(reply.as_bytes()),
810            ],
811        );
812        let mut client = Client::new(port);
813        assert_eq!(client.read_input_registers(248, 0, 1), Ok(vec![2301]));
814    }
815
816    #[test]
817    fn a_stale_reply_is_dropped_before_the_next_request() {
818        let (gateway, device) = SerialPort::pair(line());
819        device
820            .write(
821                Pdu::read_holding_registers_reply(&[1])
822                    .unwrap()
823                    .to_adu(17)
824                    .as_bytes(),
825            )
826            .unwrap();
827        let mut client = Client::new(gateway).with_response_timeout(Duration::from_millis(5));
828        assert_eq!(
829            client.read_holding_registers(17, 107, 1),
830            Err(ClientError::Timeout {
831                unit: 17,
832                received: 0
833            })
834        );
835    }
836}
pamoja_modbus/client.rs

pamoja_modbus/
client.rs
