From 11b17475f17d53e38bb2d4b40495ebf00c0323c8 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Wed, 21 Oct 2026 23:05:23 -0700 Subject: [PATCH] Bind a prepared response to the slot that owns its bytes A Prepared response was a Copy header that any code could pair with whatever objects a slot held, so its validity rested on nothing having built in that slot since, and on publishing it into the right slot. Now a slot prepares its own responses: build and empty return values that mutably borrow the slot until the response is published there or dropped. A header can only be published with the objects of the slot that prepared it, and nothing else can build in that slot meanwhile. To hold such a borrow while the session runs its handlers, the slots move out of the session into a Responder with the transport writer. Broadcast handling no longer reaches the control handler that takes the solicited slot. --- dnp3/src/outstation/control/collection.rs | 4 +- dnp3/src/outstation/response.rs | 142 +++-- dnp3/src/outstation/session.rs | 688 +++++++++++----------- dnp3/src/outstation/task.rs | 19 +- 4 files changed, 463 insertions(+), 390 deletions(-) diff --git a/dnp3/src/outstation/control/collection.rs b/dnp3/src/outstation/control/collection.rs index 7a27b7e6d..8980a8d98 100644 --- a/dnp3/src/outstation/control/collection.rs +++ b/dnp3/src/outstation/control/collection.rs @@ -21,10 +21,10 @@ impl<'a> ControlTransaction<'a> { pub(crate) async fn execute( handler: &'a mut dyn ControlHandler, database: &mut DatabaseHandle, - mut func: F, + func: F, ) -> R where - F: FnMut(&mut Self, &mut DatabaseHandle) -> R, + F: FnOnce(&mut Self, &mut DatabaseHandle) -> R, { let mut tx = ControlTransaction { started: false, diff --git a/dnp3/src/outstation/response.rs b/dnp3/src/outstation/response.rs index d84e25b78..7d474b834 100644 --- a/dnp3/src/outstation/response.rs +++ b/dnp3/src/outstation/response.rs @@ -9,6 +9,11 @@ use crate::app::*; /// response, header and objects, so a resend is always the bytes of the response it repeats. No /// caller supplies a length. The slot records the length itself, from the cursor it wrote with. /// +/// A response is prepared in the slot that sends it: [`ResponseSlot::build`] and +/// [`ResponseSlot::empty`] borrow the slot until the [`Prepared`] response is published or +/// dropped. So a header can only be published with the objects of the slot it was prepared in, +/// and nothing else writes to that slot in between. +/// /// The master doesn't need this. It sends each request or confirm from the cursor that just wrote /// it, and never resends one. A retried task formats a new request. pub(crate) struct ResponseSlot { @@ -19,35 +24,44 @@ pub(crate) struct ResponseSlot { header: Option, } -/// A response a handler has prepared, ready to publish -#[derive(Copy, Clone)] -pub(crate) enum Prepared { - /// a response with no objects - Empty(ResponseHeader), - /// a response with the objects already in the slot, from the last successful - /// [`ResponseSlot::build`] or the response published before, and nothing has built since - Stored(ResponseHeader), +/// The objects a build left in its slot, waiting for their header: all of them, or none if the +/// build failed +pub(crate) struct Staged<'a> { + slot: &'a mut ResponseSlot, } -impl Prepared { - pub(crate) fn empty_solicited(seq: Sequence, iin: Iin) -> Self { - Self::Empty(ResponseHeader::solicited(seq, iin)) +impl<'a> Staged<'a> { + pub(crate) fn with_header(self, header: ResponseHeader) -> Prepared<'a> { + Prepared { + slot: self.slot, + header, + } } +} - pub(crate) fn parameter_error(seq: Sequence) -> Self { - Self::empty_solicited(seq, Iin::default() | Iin2::PARAMETER_ERROR) - } +/// A response prepared in a slot, ready to publish there +pub(crate) struct Prepared<'a> { + slot: &'a mut ResponseSlot, + header: ResponseHeader, +} - pub(crate) fn header(self) -> ResponseHeader { - match self { - Self::Empty(header) | Self::Stored(header) => header, - } +impl<'a> Prepared<'a> { + pub(crate) fn header(&self) -> ResponseHeader { + self.header } pub(crate) fn header_mut(&mut self) -> &mut ResponseHeader { - match self { - Self::Empty(header) | Self::Stored(header) => header, + &mut self.header + } + + /// Publish the response in its slot and return the bytes to send + pub(crate) fn publish(self) -> &'a [u8] { + let Self { slot, header } = self; + for (dest, src) in slot.bytes.iter_mut().zip(header.to_bytes()) { + *dest = src; } + slot.header = Some(header); + slot.view() } } @@ -60,14 +74,22 @@ impl ResponseSlot { } } - /// Write objects after the header. This doesn't publish them, and the response published - /// before is gone. + /// Write objects after the header, for a response to prepare from the result. The response + /// published before is gone. pub(crate) fn build( &mut self, write: impl FnOnce(&mut WriteCursor) -> Result, - ) -> Result { + ) -> (Result, Staged<'_>) { self.header = None; self.len = ResponseHeader::LENGTH; + let result = self.write_objects(write); + (result, Staged { slot: self }) + } + + fn write_objects( + &mut self, + write: impl FnOnce(&mut WriteCursor) -> Result, + ) -> Result { let mut cursor = WriteCursor::new(&mut self.bytes); cursor.skip(ResponseHeader::LENGTH)?; let result = write(&mut cursor)?; @@ -75,20 +97,11 @@ impl ResponseSlot { Ok(result) } - /// Publish a response and return the bytes to send - pub(crate) fn publish(&mut self, response: Prepared) -> &[u8] { - let header = match response { - Prepared::Empty(header) => { - self.len = ResponseHeader::LENGTH; - header - } - Prepared::Stored(header) => header, - }; - for (dest, src) in self.bytes.iter_mut().zip(header.to_bytes()) { - *dest = src; - } - self.header = Some(header); - self.view() + /// Prepare a response with no objects. The response published before is gone. + pub(crate) fn empty(&mut self, header: ResponseHeader) -> Prepared<'_> { + self.header = None; + self.len = ResponseHeader::LENGTH; + Prepared { slot: self, header } } /// The published response, to send again exactly as it was @@ -125,17 +138,18 @@ mod tests { #[test] fn builds_are_not_published() { let mut slot = ResponseSlot::new(BufferSize::min()); - slot.publish(Prepared::Empty(header(1))); + slot.empty(header(1)).publish(); - slot.build(|cursor| cursor.write_u8(0xAA)).unwrap(); + let (result, _) = slot.build(|cursor| cursor.write_u8(0xAA)); + result.unwrap(); assert!(slot.ready().is_none()); - slot.publish(Prepared::Empty(header(2))); - slot.build(|cursor| { + slot.empty(header(2)).publish(); + let (result, _) = slot.build(|cursor| { cursor.write_u8(0xAA)?; cursor.write_bytes(&[0; 300]) - }) - .unwrap_err(); + }); + result.unwrap_err(); assert!(slot.ready().is_none()); assert!(slot.header().is_none()); } @@ -143,30 +157,56 @@ mod tests { #[test] fn publishes_the_header_and_objects() { let mut slot = ResponseSlot::new(BufferSize::min()); - slot.build(|cursor| cursor.write_bytes(&[0xAA, 0xBB])) - .unwrap(); + let (result, staged) = slot.build(|cursor| cursor.write_bytes(&[0xAA, 0xBB])); + result.unwrap(); let expected = [0xC1, 0x81, 0x00, 0x00, 0xAA, 0xBB]; - assert_eq!(slot.publish(Prepared::Stored(header(1))), expected); + assert_eq!(staged.with_header(header(1)).publish(), expected); + assert_eq!(slot.ready(), Some(expected.as_slice())); + } + + #[test] + fn failed_build_leaves_no_objects() { + let mut slot = ResponseSlot::new(BufferSize::min()); + let (result, staged) = slot.build(|cursor| { + cursor.write_u8(0xAA)?; + cursor.write_bytes(&[0; 300]) + }); + result.unwrap_err(); + + let expected = [0xC1, 0x81, 0x00, 0x00]; + assert_eq!(staged.with_header(header(1)).publish(), expected); assert_eq!(slot.ready(), Some(expected.as_slice())); } #[test] fn empty_response_drops_the_previous_objects() { let mut slot = ResponseSlot::new(BufferSize::min()); - slot.build(|cursor| cursor.write_bytes(&[0xAA, 0xBB])) - .unwrap(); - slot.publish(Prepared::Stored(header(1))); + let (result, staged) = slot.build(|cursor| cursor.write_bytes(&[0xAA, 0xBB])); + result.unwrap(); + staged.with_header(header(1)).publish(); let expected = [0xC2, 0x81, 0x00, 0x00]; - assert_eq!(slot.publish(Prepared::Empty(header(2))), expected); + assert_eq!(slot.empty(header(2)).publish(), expected); assert_eq!(slot.ready(), Some(expected.as_slice())); } + #[test] + fn dropped_empty_response_leaves_nothing_published() { + let mut slot = ResponseSlot::new(BufferSize::min()); + slot.empty(header(1)).publish(); + + { + let _unpublished = slot.empty(header(2)); + } + assert!(slot.ready().is_none()); + assert!(slot.header().is_none()); + } + #[test] fn clear_unpublishes() { let mut slot = ResponseSlot::new(BufferSize::min()); - slot.publish(Prepared::Empty(header(1))); + slot.empty(header(1)).publish(); slot.clear(); assert!(slot.ready().is_none()); assert!(slot.header().is_none()); diff --git a/dnp3/src/outstation/session.rs b/dnp3/src/outstation/session.rs index 0b135e053..10b13a168 100644 --- a/dnp3/src/outstation/session.rs +++ b/dnp3/src/outstation/session.rs @@ -27,7 +27,7 @@ use crate::outstation::control::collection::{ use crate::outstation::control::select::SelectState; use crate::outstation::database::DatabaseHandle; use crate::outstation::deferred::DeferredRead; -use crate::outstation::response::{Prepared, ResponseSlot}; +use crate::outstation::response::{Prepared, ResponseSlot, Staged}; use crate::outstation::task::{ConfigurationChange, OutstationMessage}; use crate::outstation::traits::*; use crate::transport::{ @@ -62,18 +62,6 @@ enum ControlRequest { selection: Option, }, DirectOperate, - DirectOperateNoAck, -} - -impl ControlRequest { - fn expects_response(self) -> bool { - match self { - ControlRequest::Select { .. } - | ControlRequest::Operate { .. } - | ControlRequest::DirectOperate => true, - ControlRequest::DirectOperateNoAck => false, - } - } } #[derive(Copy, Clone, PartialEq)] @@ -123,6 +111,11 @@ impl LastValidRequest { } } +/// The header of a solicited response that reports IIN2.2 +fn parameter_error(seq: Sequence) -> ResponseHeader { + ResponseHeader::solicited(seq, Iin::default() | Iin2::PARAMETER_ERROR) +} + pub(crate) struct SessionConfig { decode_level: DecodeLevel, confirm_timeout: Timeout, @@ -135,11 +128,6 @@ pub(crate) struct SessionConfig { keep_alive_timeout: Option, max_controls_per_request: Option, } -pub(crate) struct SessionParameters { - max_read_headers_per_request: u16, - sol_tx_buffer_size: BufferSize<249, 2048>, - unsol_tx_buffer_size: BufferSize<249, 2048>, -} impl From for SessionConfig { fn from(config: OutstationConfig) -> Self { SessionConfig { @@ -157,16 +145,6 @@ impl From for SessionConfig { } } -impl From for SessionParameters { - fn from(x: OutstationConfig) -> Self { - SessionParameters { - max_read_headers_per_request: x.read_header_limit(), - sol_tx_buffer_size: x.solicited_buffer_size, - unsol_tx_buffer_size: x.unsolicited_buffer_size, - } - } -} - #[derive(Copy, Clone)] enum UnsolicitedState { /// need to perform NULL unsolicited @@ -276,12 +254,36 @@ impl SessionState { } } -pub(crate) struct OutstationSession { - enabled: Enabled, - messages: Receiver, +/// What the session answers the master with: the transport writer, and a slot for each kind of +/// response, which keeps the response published last to send again. The slots live outside the +/// session, so a response prepared in one can be held while the session runs its handlers. +pub(crate) struct Responder { + writer: TransportWriter, /// the response to the last valid request, if it had one sol: ResponseSlot, unsol: ResponseSlot, +} + +impl Responder { + pub(crate) fn new(writer: TransportWriter, config: &OutstationConfig) -> Self { + Self { + writer, + sol: ResponseSlot::new(config.solicited_buffer_size), + unsol: ResponseSlot::new(config.unsolicited_buffer_size), + } + } + + /// reset the writer and drop both responses between communication sessions + pub(crate) fn reset(&mut self) { + self.writer.reset(); + self.sol.clear(); + self.unsol.clear(); + } +} + +pub(crate) struct OutstationSession { + enabled: Enabled, + messages: Receiver, config: SessionConfig, destination: FragmentAddr, state: SessionState, @@ -428,7 +430,7 @@ impl OutstationSession { messages: Receiver, destination: FragmentAddr, config: SessionConfig, - param: SessionParameters, + max_read_headers_per_request: u16, application: Box, information: Box, control_handler: Box, @@ -441,9 +443,7 @@ impl OutstationSession { enabled: initial_state, messages, config, - sol: ResponseSlot::new(param.sol_tx_buffer_size), - unsol: ResponseSlot::new(param.unsol_tx_buffer_size), - state: SessionState::new(param.max_read_headers_per_request), + state: SessionState::new(max_read_headers_per_request), application, info: information, control_handler, @@ -486,11 +486,11 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> RunError { loop { - if let Err(err) = self.run_idle_state(io, reader, writer, database).await { + if let Err(err) = self.run_idle_state(io, reader, out, database).await { // teardown (including self.state.reset()) is handled by SessionTeardown // in OutstationTask::run, so it also covers a cancelled run future return err; @@ -501,15 +501,13 @@ impl OutstationSession { /// reset the per-connection session state between communication sessions pub(crate) fn reset_session_state(&mut self) { self.state.reset(); - self.sol.clear(); - self.unsol.clear(); } async fn write_unsolicited( &mut self, io: &mut PhysLayer, writer: &mut TransportWriter, - mut response: Prepared, + mut response: Prepared<'_>, database: &DatabaseHandle, ) -> Result { response.header_mut().iin |= self.get_response_iin(database); @@ -517,25 +515,26 @@ impl OutstationSession { .broadcast .report(response.header_mut(), ResponseKind::Unsolicited); + let header = response.header(); writer .write( io, self.config.decode_level, self.destination, - self.unsol.publish(response), + response.publish(), ) .await?; - Ok(response.header()) + Ok(header) } async fn repeat_unsolicited( &mut self, io: &mut PhysLayer, - writer: &mut TransportWriter, + out: &mut Responder, ) -> Result<(), LinkError> { - if let Some(response) = self.unsol.ready() { - writer + if let Some(response) = out.unsol.ready() { + out.writer .write(io, self.config.decode_level, self.destination, response) .await?; } @@ -547,21 +546,17 @@ impl OutstationSession { io: &mut PhysLayer, writer: &mut TransportWriter, respond_to: FragmentAddr, - mut response: Prepared, + mut response: Prepared<'_>, database: &DatabaseHandle, ) -> Result { self.finish_solicited_header(response.header_mut(), database); + let header = response.header(); writer - .write( - io, - self.config.decode_level, - respond_to, - self.sol.publish(response), - ) + .write(io, self.config.decode_level, respond_to, response.publish()) .await?; - Ok(response.header()) + Ok(header) } /// Send a response with no objects that doesn't replace the response to the last valid request @@ -594,39 +589,38 @@ impl OutstationSession { .then(|| ResponseSeries::new(header.control.seq, header.control.fin)) } - /// Answer a request that becomes the last valid request, publishing its response if it has - /// one. Returns the confirm that the response waits for. + /// Answer a request that becomes the last valid request, publishing its response. Returns the + /// confirm that the response waits for. async fn respond( &mut self, io: &mut PhysLayer, writer: &mut TransportWriter, respond_to: FragmentAddr, request: LastValidRequest, - response: Option, + response: Prepared<'_>, database: &DatabaseHandle, ) -> Result, LinkError> { - let confirm = match response { - Some(response) => Self::awaited_confirm( - self.write_solicited(io, writer, respond_to, response, database) - .await?, - ), - None => { - self.sol.clear(); - None - } - }; + let header = self + .write_solicited(io, writer, respond_to, response, database) + .await?; + self.state.last_valid_request = Some(request); + Ok(Self::awaited_confirm(header)) + } + + /// Accept a request without a response as the last valid request, so its repeat gets none + fn respond_with_nothing(&mut self, sol: &mut ResponseSlot, request: LastValidRequest) { + sol.clear(); self.state.last_valid_request = Some(request); - Ok(confirm) } async fn repeat_solicited( &mut self, io: &mut PhysLayer, respond_to: FragmentAddr, - writer: &mut TransportWriter, + out: &mut Responder, ) -> Result<(), LinkError> { - if let Some(response) = self.sol.ready() { - writer + if let Some(response) = out.sol.ready() { + out.writer .write(io, self.config.decode_level, respond_to, response) .await?; } @@ -637,22 +631,21 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result<(), RunError> { // handle a request fragment if present - self.handle_one_request_from_idle(io, reader, writer, database) + self.handle_one_request_from_idle(io, reader, out, database) .await?; // check to see if we should perform unsolicited - let next_action = self.check_unsolicited(io, reader, writer, database).await?; + let next_action = self.check_unsolicited(io, reader, out, database).await?; // handle a deferred read request if it was produced during unsolicited - self.handle_deferred_read(io, reader, writer, database) - .await?; + self.handle_deferred_read(io, reader, out, database).await?; // check to see if we should perform a link status check - self.check_link_status(io, writer).await?; + self.check_link_status(io, &mut out.writer).await?; let next_action = next_action.select_earliest(self.next_link_status); @@ -678,7 +671,7 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result { if self.config.unsolicited.is_disabled() { @@ -689,7 +682,7 @@ impl OutstationSession { UnsolicitedState::NullRequired => { // perform NULL unsolicited match self - .perform_null_unsolicited(io, reader, writer, database) + .perform_null_unsolicited(io, reader, out, database) .await? { UnsolicitedResult::Timeout | UnsolicitedResult::ReturnToIdle => { @@ -713,7 +706,7 @@ impl OutstationSession { // perform regular unsolicited match self - .maybe_perform_unsolicited(io, reader, writer, database) + .maybe_perform_unsolicited(io, reader, out, database) .await? { None => { @@ -776,7 +769,7 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result { let header = ResponseHeader::new( @@ -784,47 +777,49 @@ impl OutstationSession { ResponseFunction::UnsolicitedResponse, Iin::default(), ); - self.perform_unsolicited_response_series( - database, - Prepared::Empty(header), - true, - io, - reader, - writer, - ) - .await + let seq = self + .write_unsolicited(io, &mut out.writer, out.unsol.empty(header), database) + .await? + .control + .seq; + self.unsolicited_confirm_wait(database, seq, true, io, reader, out) + .await } async fn maybe_perform_unsolicited( &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result, RunError> { if !self.state.enabled_unsolicited_classes.any() { return Ok(None); } - match self.write_unsolicited_data(database) { - None => Ok(None), - Some(res) => { - let result = self - .perform_unsolicited_response_series(database, res, false, io, reader, writer) - .await?; - Ok(Some(result)) - } - } + let Some(response) = self.write_unsolicited_data(&mut out.unsol, database) else { + return Ok(None); + }; + let seq = self + .write_unsolicited(io, &mut out.writer, response, database) + .await? + .control + .seq; + let result = self + .unsolicited_confirm_wait(database, seq, false, io, reader, out) + .await?; + Ok(Some(result)) } - fn write_unsolicited_data(&mut self, database: &mut DatabaseHandle) -> Option { + fn write_unsolicited_data<'a>( + &mut self, + unsol: &'a mut ResponseSlot, + database: &mut DatabaseHandle, + ) -> Option> { let classes = self.state.enabled_unsolicited_classes; - let count = self - .unsol - .build(|cursor| Ok(database.write_unsolicited(classes, cursor))) - .unwrap_or_default(); + let (count, staged) = unsol.build(|cursor| Ok(database.write_unsolicited(classes, cursor))); - if count == 0 { + if count.unwrap_or_default() == 0 { return None; } @@ -834,24 +829,20 @@ impl OutstationSession { ResponseFunction::UnsolicitedResponse, Iin::default(), ); - Some(Prepared::Stored(header)) + Some(staged.with_header(header)) } - async fn perform_unsolicited_response_series( + /// Wait for the confirm of the unsolicited response just sent with `seq`, sending it again + /// when the wait times out + async fn unsolicited_confirm_wait( &mut self, database: &mut DatabaseHandle, - response: Prepared, + seq: Sequence, is_null: bool, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, ) -> Result { - let seq = self - .write_unsolicited(io, writer, response, database) - .await? - .control - .seq; - // enter unsolicited confirm wait state self.info.enter_unsolicited_confirm_wait(seq); @@ -875,7 +866,7 @@ impl OutstationSession { loop { match self - .wait_for_unsolicited_confirm(&mut wait, deadline, io, reader, writer, database) + .wait_for_unsolicited_confirm(&mut wait, deadline, io, reader, out, database) .instrument(tracing::info_span!("UnsolConfirmWait", "seq" = seq.value())) .await? { @@ -898,7 +889,7 @@ impl OutstationSession { } // perform a retry - self.repeat_unsolicited(io, writer).await?; + self.repeat_unsolicited(io, out).await?; // update the deadline deadline = self.new_confirm_deadline(); @@ -913,7 +904,7 @@ impl OutstationSession { deadline: tokio::time::Instant, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result { if let TimeoutStatus::Yes = self.read_until(io, reader, deadline).await? { @@ -928,7 +919,7 @@ impl OutstationSession { // solicited confirm wait. Its IIN2.0 answer isn't the response to a valid request, // so the outstation doesn't wait for that answer's confirm wait.solicited = None; - self.answer_unknown_function(io, writer, from, seq, database) + self.answer_unknown_function(io, &mut out.writer, from, seq, database) .await?; return Ok(UnsolicitedWaitResult::ReadNext); } @@ -945,7 +936,7 @@ impl OutstationSession { // Table 6-1 rows 24-29: a request ends the wait for a solicited confirm, and an answer that // asks for one starts it again let confirm = self - .accept(io, writer, database, received, selection, Reads::Defer) + .accept(io, out, database, received, selection, Reads::Defer) .await?; wait.solicited = confirm.map(|series| ExpectedConfirm { ecsn: series.ecsn, @@ -1068,26 +1059,26 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result<(), RunError> { if let Some(x) = self.state.deferred_read.select(database) { tracing::info!("handling deferred READ request"); - let response = self.format_read_response(database, true, x.seq, x.iin2); + let response = Self::format_read_response(&mut out.sol, database, true, x.seq, x.iin2); let confirm = self .respond( io, - writer, + &mut out.writer, x.info.addr, LastValidRequest::new(x.seq, x.hash), - Some(response), + response, database, ) .await?; if let Some(series) = confirm { // enter the solicited confirm wait state - self.sol_confirm_wait(io, reader, writer, database, series) + self.sol_confirm_wait(io, reader, out, database, series) .instrument(tracing::info_span!( "SolConfirmWait", "ecsn" = series.ecsn.value() @@ -1104,14 +1095,14 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, ) -> Result<(), RunError> { let guard = self.pop_request(reader); let confirm = match self.receive(&guard) { Intake::Nothing => None, Intake::UnknownFunction(from, seq) => { - self.answer_unknown_function(io, writer, from, seq, database) + self.answer_unknown_function(io, &mut out.writer, from, seq, database) .await?; None } @@ -1131,7 +1122,7 @@ impl OutstationSession { Intake::Request(received) => { let selection = self.take_selection(&received.fragment); self.info.process_request_from_idle(received.request.header); - self.accept(io, writer, database, received, selection, Reads::Answer) + self.accept(io, out, database, received, selection, Reads::Answer) .await? } }; @@ -1139,7 +1130,7 @@ impl OutstationSession { if let Some(series) = confirm { // enter the solicited confirm wait state - self.sol_confirm_wait(io, reader, writer, database, series) + self.sol_confirm_wait(io, reader, out, database, series) .instrument(tracing::info_span!( "SolConfirmWait", "ecsn" = series.ecsn.value() @@ -1199,7 +1190,7 @@ impl OutstationSession { async fn accept( &mut self, io: &mut PhysLayer, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, received: Received<'_>, selection: Option, @@ -1224,13 +1215,14 @@ impl OutstationSession { // response to a repeated READ. We deliberately answer it as a new READ // instead: the fresh response carries current values, and events that were // never confirmed are still buffered, so they are reported again - let response = self.format_first_read_response(database, seq, objects); + let response = + Self::format_first_read_response(&mut out.sol, database, seq, objects); self.respond( io, - writer, + &mut out.writer, info.addr, LastValidRequest::new(seq, hash), - Some(response), + response, database, ) .await @@ -1244,43 +1236,45 @@ impl OutstationSession { FragmentType::MalformedRequest(hash, err) => { // IEEE 1815-2012 clause 6.1: a request whose objects don't parse is still the last // request, answered with IIN2 - let response = Prepared::empty_solicited(seq, Iin::default() | Iin2::from(err)); + let header = ResponseHeader::solicited(seq, Iin::default() | Iin2::from(err)); self.respond( io, - writer, + &mut out.writer, info.addr, LastValidRequest::new(seq, hash), - Some(response), + out.sol.empty(header), database, ) .await } FragmentType::NewNonRead(hash, objects) => { + let last = LastValidRequest::new(seq, hash); let response = self .handle_non_read( + &mut out.sol, database, request.header.function, - seq, - hash, + last, selection, objects, ) .await; - self.respond( - io, - writer, - info.addr, - LastValidRequest::new(seq, hash), - response, - database, - ) - .await + match response { + Some(response) => { + self.respond(io, &mut out.writer, info.addr, last, response, database) + .await + } + None => { + self.respond_with_nothing(&mut out.sol, last); + Ok(None) + } + } } FragmentType::RepeatNonRead(_) => { // IEEE 1815-2012 Table 6-1: send the same response, byte for byte, which waits for // its confirm again if it asked for one - self.repeat_solicited(io, info.addr, writer).await?; - Ok(self.sol.header().and_then(Self::awaited_confirm)) + self.repeat_solicited(io, info.addr, out).await?; + Ok(out.sol.header().and_then(Self::awaited_confirm)) } FragmentType::Broadcast(mode) => { self.process_broadcast(database, mode, request).await; @@ -1328,130 +1322,142 @@ impl OutstationSession { } } - fn format_first_read_response( - &mut self, + fn format_first_read_response<'a>( + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, seq: Sequence, object_headers: HeaderCollection, - ) -> Prepared { + ) -> Prepared<'a> { let iin2 = database.select(&object_headers); - self.format_read_response(database, true, seq, iin2) + Self::format_read_response(sol, database, true, seq, iin2) } - fn format_read_response( - &mut self, + fn format_read_response<'a>( + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, fir: bool, seq: Sequence, iin2: Iin2, - ) -> Prepared { - let info = match self - .sol - .build(|cursor| Ok(database.write_response_headers(cursor))) - { - Ok(info) => info, + ) -> Prepared<'a> { + let (result, staged) = sol.build(|cursor| Ok(database.write_response_headers(cursor))); + let header = match result { + Ok(info) => ResponseHeader::new( + ControlField::response(seq, fir, info.complete, info.need_confirm()), + ResponseFunction::Response, + Iin::default() | iin2, + ), Err(err) => { tracing::error!("unable to write a READ response: {err:?}"); - return Prepared::parameter_error(seq); + parameter_error(seq) } }; - - let header = ResponseHeader::new( - ControlField::response(seq, fir, info.complete, info.need_confirm()), - ResponseFunction::Response, - Iin::default() | iin2, - ); - Prepared::Stored(header) + staged.with_header(header) } - async fn handle_non_read( + async fn handle_non_read<'a>( &mut self, + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, function: FunctionCode, - seq: Sequence, - request_hash: u64, + request: LastValidRequest, selection: Option, object_headers: HeaderCollection<'_>, - ) -> Option { + ) -> Option> { + let LastValidRequest { seq, request_hash } = request; let mut result = match function { - FunctionCode::Write => Some(self.handle_write(seq, object_headers, database).await), + FunctionCode::Write => { + let header = self.handle_write(seq, object_headers, database).await; + Some(sol.empty(header)) + } // these function don't process objects - FunctionCode::DelayMeasure => Some(self.handle_delay_measure(seq)), - FunctionCode::RecordCurrentTime => Some(self.handle_record_current_time(seq)), + FunctionCode::DelayMeasure => Some(self.handle_delay_measure(sol, seq)), + FunctionCode::RecordCurrentTime => { + Some(sol.empty(self.handle_record_current_time(seq))) + } FunctionCode::ColdRestart => { let delay = self.application.cold_restart(); - Some(self.handle_restart(seq, delay)) + Some(Self::handle_restart(sol, seq, delay)) } FunctionCode::WarmRestart => { let delay = self.application.warm_restart(); - Some(self.handle_restart(seq, delay)) + Some(Self::handle_restart(sol, seq, delay)) } // controls - FunctionCode::Select => { + FunctionCode::Select => Some( self.handle_controls( + sol, ControlRequest::Select { request_hash }, database, seq, object_headers, ) - .await - } - FunctionCode::Operate => { + .await, + ), + FunctionCode::Operate => Some( self.handle_controls( + sol, ControlRequest::Operate { selection }, database, seq, object_headers, ) - .await - } - FunctionCode::DirectOperate => { - self.handle_controls(ControlRequest::DirectOperate, database, seq, object_headers) - .await - } - FunctionCode::DirectOperateNoResponse => { + .await, + ), + FunctionCode::DirectOperate => Some( self.handle_controls( - ControlRequest::DirectOperateNoAck, + sol, + ControlRequest::DirectOperate, database, seq, object_headers, ) - .await - } - FunctionCode::ImmediateFreeze => { - Some(self.handle_freeze(database, seq, object_headers, FreezeType::ImmediateFreeze)) + .await, + ), + FunctionCode::DirectOperateNoResponse => { + self.handle_direct_operate_no_ack(database, object_headers) + .await; + None } + FunctionCode::ImmediateFreeze => Some(sol.empty(self.handle_freeze( + database, + seq, + object_headers, + FreezeType::ImmediateFreeze, + ))), FunctionCode::ImmediateFreezeNoResponse => { self.handle_freeze(database, seq, object_headers, FreezeType::ImmediateFreeze); None } - FunctionCode::FreezeClear => { - Some(self.handle_freeze(database, seq, object_headers, FreezeType::FreezeAndClear)) - } + FunctionCode::FreezeClear => Some(sol.empty(self.handle_freeze( + database, + seq, + object_headers, + FreezeType::FreezeAndClear, + ))), FunctionCode::FreezeClearNoResponse => { self.handle_freeze(database, seq, object_headers, FreezeType::FreezeAndClear); None } FunctionCode::FreezeAtTime => { - Some(self.handle_freeze_at_time(database, seq, object_headers)) + Some(sol.empty(self.handle_freeze_at_time(database, seq, object_headers))) } FunctionCode::FreezeAtTimeNoResponse => { self.handle_freeze_at_time(database, seq, object_headers); None } - FunctionCode::EnableUnsolicited => { - Some(self.handle_enable_or_disable_unsolicited(true, seq, object_headers)) - } - FunctionCode::DisableUnsolicited => { - Some(self.handle_enable_or_disable_unsolicited(false, seq, object_headers)) - } + FunctionCode::EnableUnsolicited => Some( + sol.empty(self.handle_enable_or_disable_unsolicited(true, seq, object_headers)), + ), + FunctionCode::DisableUnsolicited => Some( + sol.empty(self.handle_enable_or_disable_unsolicited(false, seq, object_headers)), + ), _ => { tracing::warn!("unsupported function code: {:?}", function); - Some(Prepared::empty_solicited( + Some(sol.empty(ResponseHeader::solicited( seq, Iin::default() | Iin2::NO_FUNC_CODE_SUPPORT, - )) + ))) } }; @@ -1480,7 +1486,7 @@ impl OutstationSession { seq: Sequence, object_headers: HeaderCollection<'_>, database: &DatabaseHandle, - ) -> Prepared { + ) -> ResponseHeader { let mut iin2 = Iin2::default(); let mut empty = true; @@ -1493,7 +1499,7 @@ impl OutstationSession { tracing::warn!("empty WRITE request"); } - Prepared::empty_solicited(seq, Iin::default() | iin2) + ResponseHeader::solicited(seq, Iin::default() | iin2) } fn handle_write_iin(&mut self, bits: BitSequence) -> Iin2 { @@ -1688,40 +1694,51 @@ impl OutstationSession { } } - fn handle_delay_measure(&mut self, seq: Sequence) -> Prepared { + fn handle_delay_measure<'a>( + &mut self, + sol: &'a mut ResponseSlot, + seq: Sequence, + ) -> Prepared<'a> { let g52v2 = Group52Var2 { time: self.application.get_processing_delay_ms(), }; - self.single_object_response(seq, g52v2) + Self::single_object_response(sol, seq, g52v2) } /// Respond with a single object, or with IIN2.2 if it can't be written - fn single_object_response(&mut self, seq: Sequence, item: V) -> Prepared + fn single_object_response(sol: &mut ResponseSlot, seq: Sequence, item: V) -> Prepared<'_> where V: FixedSizeVariation, { - match self - .sol - .build(|cursor| HeaderWriter::new(cursor).write_count_of_one(item)) - { - Ok(()) => Prepared::Stored(ResponseHeader::solicited(seq, Iin::default())), + let (result, staged) = + sol.build(|cursor| HeaderWriter::new(cursor).write_count_of_one(item)); + let header = match result { + Ok(()) => ResponseHeader::solicited(seq, Iin::default()), Err(err) => { tracing::error!("unable to write {item} to the response buffer: {err:?}"); - Prepared::parameter_error(seq) + parameter_error(seq) } - } + }; + staged.with_header(header) } - fn handle_record_current_time(&mut self, seq: Sequence) -> Prepared { + fn handle_record_current_time(&mut self, seq: Sequence) -> ResponseHeader { self.state.last_recorded_time = Some(tokio::time::Instant::now()); - Prepared::empty_solicited(seq, Iin::default()) + ResponseHeader::solicited(seq, Iin::default()) } - fn handle_restart(&mut self, seq: Sequence, delay: Option) -> Prepared { + fn handle_restart( + sol: &mut ResponseSlot, + seq: Sequence, + delay: Option, + ) -> Prepared<'_> { let delay = match delay { None => { - return Prepared::empty_solicited(seq, Iin::default() | Iin2::NO_FUNC_CODE_SUPPORT); + return sol.empty(ResponseHeader::solicited( + seq, + Iin::default() | Iin2::NO_FUNC_CODE_SUPPORT, + )); } Some(x) => x, }; @@ -1729,60 +1746,48 @@ impl OutstationSession { // respond with the delay match delay { RestartDelay::Seconds(value) => { - self.single_object_response(seq, Group52Var1 { time: value }) + Self::single_object_response(sol, seq, Group52Var1 { time: value }) } RestartDelay::Milliseconds(value) => { - self.single_object_response(seq, Group52Var2 { time: value }) + Self::single_object_response(sol, seq, Group52Var2 { time: value }) } } } - async fn handle_direct_operate( - &mut self, - database: &mut DatabaseHandle, - seq: Sequence, - controls: ControlCollection<'_>, - ) -> Prepared { - let result = self - .execute_controls( - database, - ControlMode::Operate(OperateType::DirectOperate), - &controls, - ) - .await; - - Self::control_response(seq, result) - } - /// Execute the controls and build the response that echoes them, returning their overall /// status - async fn execute_controls( + async fn execute_controls<'a>( &mut self, + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, mode: ControlMode, controls: &ControlCollection<'_>, - ) -> Result { + ) -> (Result, Staged<'a>) { let max_controls_per_request = self.config.max_controls_per_request; - ControlTransaction::execute(self.control_handler.borrow_mut(), database, |tx, db| { - self.sol.build(|cursor| { - controls.execute_with_response( - cursor, - mode, - tx, - db, - ControlLimit::new(max_controls_per_request), - ) - }) - }) + ControlTransaction::execute( + self.control_handler.borrow_mut(), + database, + move |tx, db| { + sol.build(|cursor| { + controls.execute_with_response( + cursor, + mode, + tx, + db, + ControlLimit::new(max_controls_per_request), + ) + }) + }, + ) .await } /// Echo the controls into an unpublished response, without executing them, to check that the /// real response fits. The echo is the same size whatever the status. - fn control_response_fits(&mut self, controls: &ControlCollection) -> bool { - self.sol - .build(|cursor| controls.respond_with_status(cursor, CommandStatus::Success)) - .is_ok() + fn control_response_fits(sol: &mut ResponseSlot, controls: &ControlCollection) -> bool { + let (result, _) = + sol.build(|cursor| controls.respond_with_status(cursor, CommandStatus::Success)); + result.is_ok() } /// `result` is the overall status of the controls built into the response, or the error @@ -1790,22 +1795,22 @@ impl OutstationSession { fn control_response( seq: Sequence, result: Result, - ) -> Prepared { - let status = match result { - Ok(x) => x, + staged: Staged<'_>, + ) -> Prepared<'_> { + let header = match result { + Ok(status) => { + let mut iin = Iin::default(); + if status == CommandStatus::NotSupported { + iin |= Iin2::PARAMETER_ERROR; + } + ResponseHeader::solicited(seq, iin) + } Err(_) => { tracing::warn!("control response doesn't fit in the response buffer"); - return Prepared::parameter_error(seq); + parameter_error(seq) } }; - - let mut iin = Iin::default(); - - if status == CommandStatus::NotSupported { - iin |= Iin2::PARAMETER_ERROR; - } - - Prepared::Stored(ResponseHeader::solicited(seq, iin)) + staged.with_header(header) } fn handle_enable_or_disable_unsolicited( @@ -1813,7 +1818,7 @@ impl OutstationSession { enable: bool, seq: Sequence, object_headers: HeaderCollection, - ) -> Prepared { + ) -> ResponseHeader { fn to_string(enable: bool) -> &'static str { if enable { "ENABLE" @@ -1824,7 +1829,7 @@ impl OutstationSession { if self.config.unsolicited.is_disabled() { tracing::warn!("received {} unsolicited request, but unsolicited support is disabled by configuration", to_string(enable)); - return Prepared::empty_solicited(seq, Iin::default() | Iin2::NO_FUNC_CODE_SUPPORT); + return ResponseHeader::solicited(seq, Iin::default() | Iin2::NO_FUNC_CODE_SUPPORT); } let mut iin2 = Iin2::default(); @@ -1852,14 +1857,34 @@ impl OutstationSession { } } - Prepared::empty_solicited(seq, Iin::default() | iin2) + ResponseHeader::solicited(seq, Iin::default() | iin2) + } + + /// The controls in a request, or `None` if it has an object that the control functions don't + /// support. A control object with an unsupported qualifier fails to parse, so it's one of + /// those. + fn controls(object_headers: HeaderCollection<'_>) -> Option> { + match ControlCollection::from(object_headers) { + Ok(controls) => Some(controls), + Err(err) => { + tracing::warn!( + "ignoring control request containing non-control object header {} - {}", + err.variation, + err.qualifier + ); + None + } + } } async fn handle_direct_operate_no_ack( &mut self, database: &mut DatabaseHandle, - controls: ControlCollection<'_>, + object_headers: HeaderCollection<'_>, ) { + let Some(controls) = Self::controls(object_headers) else { + return; + }; let max_controls_per_request = self.config.max_controls_per_request; ControlTransaction::execute(self.control_handler.borrow_mut(), database, |tx, db| { controls.operate_no_ack(tx, db, ControlLimit::new(max_controls_per_request)) @@ -1867,64 +1892,61 @@ impl OutstationSession { .await; } - async fn handle_controls( + async fn handle_controls<'a>( &mut self, + sol: &'a mut ResponseSlot, request: ControlRequest, database: &mut DatabaseHandle, seq: Sequence, object_headers: HeaderCollection<'_>, - ) -> Option { - // a control object with an unsupported qualifier fails to parse, so this is an object - // that the control functions don't support - let controls = match ControlCollection::from(object_headers) { - Err(err) => { - tracing::warn!( - "ignoring control request containing non-control object header {} - {}", - err.variation, - err.qualifier - ); - return request.expects_response().then(|| { - Prepared::empty_solicited(seq, Iin::default() | Iin2::OBJECT_UNKNOWN) - }); - } - Ok(controls) => controls, + ) -> Prepared<'a> { + let Some(controls) = Self::controls(object_headers) else { + return sol.empty(ResponseHeader::solicited( + seq, + Iin::default() | Iin2::OBJECT_UNKNOWN, + )); }; - if request.expects_response() && !self.control_response_fits(&controls) { + if !Self::control_response_fits(sol, &controls) { tracing::warn!( "ignoring control request whose response doesn't fit in the response buffer" ); - return Some(Prepared::parameter_error(seq)); + return sol.empty(parameter_error(seq)); } match request { - ControlRequest::Select { request_hash } => Some( - self.handle_select(database, seq, request_hash, controls) - .await, - ), - ControlRequest::Operate { selection } => Some( - self.handle_operate(database, seq, selection, controls) - .await, - ), - ControlRequest::DirectOperate => { - Some(self.handle_direct_operate(database, seq, controls).await) + ControlRequest::Select { request_hash } => { + self.handle_select(sol, database, seq, request_hash, controls) + .await } - ControlRequest::DirectOperateNoAck => { - self.handle_direct_operate_no_ack(database, controls).await; - None + ControlRequest::Operate { selection } => { + self.handle_operate(sol, database, seq, selection, controls) + .await + } + ControlRequest::DirectOperate => { + let (result, staged) = self + .execute_controls( + sol, + database, + ControlMode::Operate(OperateType::DirectOperate), + &controls, + ) + .await; + Self::control_response(seq, result, staged) } } } - async fn handle_select( + async fn handle_select<'a>( &mut self, + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, seq: Sequence, request_hash: u64, controls: ControlCollection<'_>, - ) -> Prepared { - let result = self - .execute_controls(database, ControlMode::Select, &controls) + ) -> Prepared<'a> { + let (result, staged) = self + .execute_controls(sol, database, ControlMode::Select, &controls) .await; // Record the select state @@ -1937,28 +1959,30 @@ impl OutstationSession { )) } - Self::control_response(seq, result) + Self::control_response(seq, result, staged) } - async fn handle_operate( + async fn handle_operate<'a>( &mut self, + sol: &'a mut ResponseSlot, database: &mut DatabaseHandle, seq: Sequence, selection: Option, controls: ControlCollection<'_>, - ) -> Prepared { + ) -> Prepared<'a> { // the OPERATE has already ended the selection, so it is used at most once let matched = selection .ok_or(CommandStatus::NoSelect) .and_then(|s| s.match_operate(self.config.select_timeout, seq, controls.hash())); // Handle each operate and write the response - let result = match matched { - Err(status) => self - .sol - .build(|cursor| controls.respond_with_status(cursor, status).map(|_| status)), + let (result, staged) = match matched { + Err(status) => { + sol.build(|cursor| controls.respond_with_status(cursor, status).map(|_| status)) + } Ok(()) => { self.execute_controls( + sol, database, ControlMode::Operate(OperateType::SelectBeforeOperate), &controls, @@ -1967,7 +1991,7 @@ impl OutstationSession { } }; - Self::control_response(seq, result) + Self::control_response(seq, result, staged) } fn handle_freeze( @@ -1976,14 +2000,14 @@ impl OutstationSession { seq: Sequence, object_headers: HeaderCollection, freeze_type: FreezeType, - ) -> Prepared { + ) -> ResponseHeader { let mut iin = Iin::default(); for header in object_headers.iter() { iin |= self.handle_freeze_header(database, freeze_type, header); } - Prepared::empty_solicited(seq, iin) + ResponseHeader::solicited(seq, iin) } fn handle_freeze_at_time( @@ -1991,7 +2015,7 @@ impl OutstationSession { database: &mut DatabaseHandle, seq: Sequence, object_headers: HeaderCollection, - ) -> Prepared { + ) -> ResponseHeader { let mut iin = Iin::default(); let mut timing: Option = None; @@ -2034,7 +2058,7 @@ impl OutstationSession { } } - Prepared::empty_solicited(seq, iin) + ResponseHeader::solicited(seq, iin) } fn handle_freeze_header( @@ -2138,8 +2162,7 @@ impl OutstationSession { BroadcastAction::Processed } FunctionCode::DirectOperateNoResponse => { - self.handle_controls(ControlRequest::DirectOperateNoAck, database, seq, objects) - .await; + self.handle_direct_operate_no_ack(database, objects).await; BroadcastAction::Processed } FunctionCode::ImmediateFreezeNoResponse => { @@ -2191,7 +2214,7 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, database: &mut DatabaseHandle, mut series: ResponseSeries, ) -> Result<(), RunError> { @@ -2199,7 +2222,7 @@ impl OutstationSession { loop { match self - .wait_for_sol_confirm(io, reader, writer, series.ecsn) + .wait_for_sol_confirm(io, reader, out, series.ecsn) .await? { Confirm::Yes(respond_to) => { @@ -2214,10 +2237,15 @@ impl OutstationSession { } // format the next response in the series series.ecsn.increment(); - let response = - self.format_read_response(database, false, series.ecsn, Iin2::default()); + let response = Self::format_read_response( + &mut out.sol, + database, + false, + series.ecsn, + Iin2::default(), + ); let header = self - .write_solicited(io, writer, respond_to, response, database) + .write_solicited(io, &mut out.writer, respond_to, response, database) .await?; match Self::awaited_confirm(header) { None => return Ok(()), @@ -2246,7 +2274,7 @@ impl OutstationSession { &mut self, io: &mut PhysLayer, reader: &mut TransportReader, - writer: &mut TransportWriter, + out: &mut Responder, ecsn: Sequence, ) -> Result { let mut deadline = self.new_confirm_deadline(); @@ -2275,7 +2303,7 @@ impl OutstationSession { return Ok(Confirm::NewRequest); } ConfirmAction::EchoLastResponse(respond_to) => { - self.repeat_solicited(io, respond_to, writer).await?; + self.repeat_solicited(io, respond_to, out).await?; // per the spec, we restart the confirm timer deadline = self.new_confirm_deadline(); } diff --git a/dnp3/src/outstation/task.rs b/dnp3/src/outstation/task.rs index 32af6c40a..9978ddad4 100644 --- a/dnp3/src/outstation/task.rs +++ b/dnp3/src/outstation/task.rs @@ -4,10 +4,10 @@ use crate::link::reader::LinkModes; use crate::link::EndpointAddress; use crate::outstation::config::*; use crate::outstation::database::DatabaseHandle; -use crate::outstation::session::OutstationSession; +use crate::outstation::session::{OutstationSession, Responder}; use crate::outstation::traits::{ControlHandler, OutstationApplication, OutstationInformation}; use crate::outstation::OutstationHandle; -use crate::transport::{FragmentAddr, RealTransport, Transport, TransportReader, TransportWriter}; +use crate::transport::{FragmentAddr, RealTransport, Transport, TransportReader}; use crate::util::phys::{PhysAddr, PhysLayer}; use crate::util::session::{Enabled, RunError, StopReason}; @@ -31,7 +31,7 @@ pub(crate) enum OutstationMessage { pub(crate) struct OutstationTask { session: OutstationSession, reader: TransportReader, - writer: TransportWriter, + responder: Responder, database: DatabaseHandle, } @@ -45,7 +45,7 @@ impl Drop for SessionTeardown<'_, T> { // re-arm, don't clear: unconfirmed events return to Unselected for re-delivery self.task.database.reset(); self.task.reader.reset(); - self.task.writer.reset(); + self.task.responder.reset(); self.task.session.reset_session_state(); } } @@ -132,13 +132,13 @@ impl OutstationTask { rx, destination, config.into(), - config.into(), + config.read_header_limit(), application, information, control_handler, ), reader, - writer, + responder: Responder::new(writer, &config), database: handle.clone(), }; ( @@ -166,7 +166,12 @@ impl OutstationTask { let guard = SessionTeardown { task: self }; let task = &mut *guard.task; // reborrow so Drop can reset after the run future releases its borrows task.session - .run(io, &mut task.reader, &mut task.writer, &mut task.database) + .run( + io, + &mut task.reader, + &mut task.responder, + &mut task.database, + ) .await }