[ul] Fix max PDU issues

ul/src/association/client.rs

@@ -14,9 +14,13 @@ use std::{
 
 use crate::{
     AeAddr, IMPLEMENTATION_CLASS_UID, IMPLEMENTATION_VERSION_NAME, association::{
-        Association, CloseSocket, NegotiatedOptions, SocketOptions, SyncAssociation, encode_pdu, private::SyncAssociationSealed, read_pdu_from_wire
+        Association, CloseSocket, NegotiatedOptions, SocketOptions, SyncAssociation, encode_pdu,
+        private::SyncAssociationSealed, read_pdu_from_wire,
     }, pdu::{
-        AbortRQSource, AssociationAC, AssociationRQ, DEFAULT_MAX_PDU, LARGE_PDU_SIZE, PDU_HEADER_SIZE, Pdu, PresentationContextNegotiated, PresentationContextProposed, PresentationContextResultReason, UserIdentity, UserIdentityType, UserVariableItem, write_pdu
+        AbortRQSource, AssociationAC, AssociationRQ, DEFAULT_MAX_PDU, LARGE_PDU_SIZE,
+        PDU_HEADER_SIZE, Pdu, PresentationContextNegotiated, PresentationContextProposed,
+        PresentationContextResultReason, UserIdentity, UserIdentityType, UserVariableItem,
+        write_pdu,
     }
 };
 use snafu::{ensure, ResultExt};
@@ -975,11 +979,27 @@ where S: CloseSocket + std::io::Read + std::io::Write,
         &self.peer_ae_title
     }
 
+    /// Retrieve the maximum PDU length
+    /// that the association acceptor is expecting to receive.
+    fn acceptor_max_pdu_length(&self) -> u32 {
+        self.acceptor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length
+    /// that the association requestor is expecting to receive.
     fn requestor_max_pdu_length(&self) -> u32 {
         self.requestor_max_pdu_length
     }
 
-    fn acceptor_max_pdu_length(&self) -> u32 {
+    /// Retrieve the maximum PDU length that this application entity
+    /// (the association requestor) is expecting to receive.
+    fn local_max_pdu_length(&self) -> u32 {
+        self.requestor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length that the peer application entity
+    /// (the association acceptor) is expecting to receive.
+    fn peer_max_pdu_length(&self) -> u32 {
         self.acceptor_max_pdu_length
     }
 
@@ -1326,12 +1346,28 @@ where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
         &self.peer_ae_title
     }
 
+    /// Retrieve the maximum PDU length
+    /// that the association acceptor is expecting to receive.
     fn acceptor_max_pdu_length(&self) -> u32 {
         self.acceptor_max_pdu_length
     }
 
+    /// Retrieve the maximum PDU length
+    /// that the association requestor is expecting to receive.
     fn requestor_max_pdu_length(&self) -> u32 {
-        self.requestor_max_pdu_length 
+        self.requestor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length that this application entity
+    /// (the association requestor) is expecting to receive.
+    fn local_max_pdu_length(&self) -> u32 {
+        self.requestor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length that the peer application entity
+    /// (the association acceptor) is expecting to receive.
+    fn peer_max_pdu_length(&self) -> u32 {
+        self.acceptor_max_pdu_length
     }
 
     fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {

ul/src/association/mod.rs

@@ -238,13 +238,27 @@ pub trait Association {
     /// admitted by the association acceptor.
     fn acceptor_max_pdu_length(&self) -> u32;
 
+    /// Retrieve the maximum PDU length
+    /// admitted by the association requestor.
+    fn requestor_max_pdu_length(&self) -> u32;
+
     /// Retrieve the maximum PDU length
     /// that this application entity is expecting to receive.
+    /// That's the same as acceptor_max_pdu_length() for
+    /// server objects, and as requestor_max_pdu_length()
+    /// for client objects.
     ///
     /// The current implementation is not required to fail
     /// and/or abort the association
     /// if a larger PDU is received.
-    fn requestor_max_pdu_length(&self) -> u32;
+    fn local_max_pdu_length(&self) -> u32;
+
+    /// Retrieve the maximum PDU length
+    /// admitted by the peer.
+    /// That's the same as requestor_max_pdu_length() for
+    /// server objects, and as acceptor_max_pdu_length()
+    /// for client objects.
+    fn peer_max_pdu_length(&self) -> u32;
 
     /// Obtain a view of the negotiated presentation contexts.
     fn presentation_contexts(&self) -> &[PresentationContextNegotiated];
@@ -406,13 +420,12 @@ pub trait SyncAssociation<S: std::io::Read + std::io::Write + CloseSocket>: priv
     /// Returns a writer which automatically
     /// splits the inner data into separate PDUs if necessary.
     fn send_pdata(&mut self, presentation_context_id: u8) -> PDataWriter<&mut S>{
-        let max_pdu_length = self.acceptor_max_pdu_length();
+        let max_pdu_length = self.peer_max_pdu_length();
         PDataWriter::new(
             self.inner_stream(),
             presentation_context_id,
             max_pdu_length,
         )
-
     }
 
     /// Prepare a P-Data reader for receiving
@@ -421,7 +434,7 @@ pub trait SyncAssociation<S: std::io::Read + std::io::Write + CloseSocket>: priv
     /// Returns a reader which automatically
     /// receives more data PDUs once the bytes collected are consumed.
     fn receive_pdata(&mut self) -> PDataReader<'_, &mut S>{
-        let max_pdu_length = self.requestor_max_pdu_length();
+        let max_pdu_length = self.local_max_pdu_length();
         let (socket, read_buffer) = self.get_mut();
         PDataReader::new(
             socket,
@@ -489,13 +502,12 @@ pub trait AsyncAssociation<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unp
     /// Returns a writer which automatically
     /// splits the inner data into separate PDUs if necessary.
     fn send_pdata(&mut self, presentation_context_id: u8) -> AsyncPDataWriter<&mut S>{
-        let max_pdu_length = self.acceptor_max_pdu_length();
+        let max_pdu_length = self.peer_max_pdu_length();
         AsyncPDataWriter::new(
             self.inner_stream(),
             presentation_context_id,
             max_pdu_length,
         )
-
     }
 
     /// Prepare a P-Data reader for receiving
@@ -504,7 +516,7 @@ pub trait AsyncAssociation<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unp
     /// Returns a reader which automatically
     /// receives more data PDUs once the bytes collected are consumed.
     fn receive_pdata(&mut self) -> PDataReader<'_, &mut S>{
-        let max_pdu_length = self.requestor_max_pdu_length();
+        let max_pdu_length = self.local_max_pdu_length();
         let (socket, read_buffer) = self.get_mut();
         PDataReader::new(
             socket,

ul/src/association/server.rs

@@ -812,17 +812,29 @@ where S: std::io::Read + std::io::Write + CloseSocket{
     }
 
     /// Retrieve the maximum PDU length
-    /// admitted by this application entity.
+    /// that the association acceptor is expecting to receive.
     fn acceptor_max_pdu_length(&self) -> u32 {
         self.acceptor_max_pdu_length
     }
 
     /// Retrieve the maximum PDU length
-    /// that the requestor is expecting to receive.
+    /// that the association requestor is expecting to receive.
     fn requestor_max_pdu_length(&self) -> u32 {
         self.requestor_max_pdu_length
     }
 
+    /// Retrieve the maximum PDU length that this application entity
+    /// (the association acceptor) is expecting to receive.
+    fn local_max_pdu_length(&self) -> u32 {
+        self.acceptor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length that the peer application entity
+    /// (the association requestor) is expecting to receive.
+    fn peer_max_pdu_length(&self) -> u32 {
+        self.requestor_max_pdu_length
+    }
+
     /// Obtain the remote DICOM node's application entity title.
     fn peer_ae_title(&self) -> &str {
         &self.client_ae_title
@@ -1076,14 +1088,30 @@ where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send{
 impl<S> Association for AsyncServerAssociation<S> 
 where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send{
 
+    /// Retrieve the maximum PDU length
+    /// that the association acceptor is expecting to receive.
     fn acceptor_max_pdu_length(&self) -> u32 {
         self.acceptor_max_pdu_length
     }
 
+    /// Retrieve the maximum PDU length
+    /// that the association requestor is expecting to receive.
     fn requestor_max_pdu_length(&self) -> u32 {
         self.requestor_max_pdu_length
     }
 
+    /// Retrieve the maximum PDU length that this application entity
+    /// (the association acceptor) is expecting to receive.
+    fn local_max_pdu_length(&self) -> u32 {
+        self.acceptor_max_pdu_length
+    }
+
+    /// Retrieve the maximum PDU length that the peer application entity
+    /// (the association requestor) is expecting to receive.
+    fn peer_max_pdu_length(&self) -> u32 {
+        self.requestor_max_pdu_length
+    }
+
     /// Obtain a view of the negotiated presentation contexts.
     fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
         &self.presentation_contexts

ul/tests/association_echo.rs

@@ -6,6 +6,10 @@ use dicom_ul::{
 };
 #[cfg(feature = "async")]
 use dicom_ul::association::AsyncServerAssociation;
+use std::io::Write;
+
+#[cfg(feature = "async")]
+use tokio::io::AsyncWriteExt;
 
 use std::net::SocketAddr;
 
@@ -109,6 +113,35 @@ fn spawn_scp(
             e => panic!("Expected SendTooLongPdu but didn't happen: {:?}", e),
         }
 
+        // Test send_pdata() fragmentation of the client; we should receive two packets
+        // First packet
+        match association.receive() {
+            Ok(Pdu::PData { data }) => {
+                assert_eq!(data.len(), 1);
+                assert_eq!(data[0].data.len(), max_server_pdu_len - PDV_HDR_LEN);
+            }
+            Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+            Err(err) => { panic!("Receive returned error {:?}", err); }
+        }
+        // Second packet
+        match association.receive() {
+            Ok(Pdu::PData { data }) => {
+                assert_eq!(data.len(), 1);
+                assert_eq!(data[0].data.len(), 2);
+            }
+            Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+            Err(err) => { panic!("Receive returned error {:?}", err); }
+        }
+        // Let the client test our send_pdata() fragmentation for us
+        {
+            // Send two more bytes than fit in a PDU
+            let filler_len = max_client_pdu_len - PDV_HDR_LEN + 2;
+            let buf = vec![0_u8; filler_len];
+            let mut sender = association.send_pdata(1);
+            // This should split the data in two packets
+            sender.write_all(&buf).expect("Error sending fragmented data");
+        }
+
         // handle one release request
         let pdu = association.receive()?;
         assert_eq!(pdu, Pdu::ReleaseRQ);
@@ -195,6 +228,35 @@ async fn spawn_scp_async(
             e => panic!("Expected SendTooLongPdu but didn't happen: {:?}", e),
         }
 
+        // Test send_pdata() fragmentation of the client; we should receive two packets
+        // First packet
+        match association.receive().await {
+            Ok(Pdu::PData { data }) => {
+                assert_eq!(data.len(), 1);
+                assert_eq!(data[0].data.len(), max_server_pdu_len - PDV_HDR_LEN);
+            }
+            Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+            Err(err) => { panic!("Receive returned error {:?}", err); }
+        }
+        // Second packet
+        match association.receive().await {
+            Ok(Pdu::PData { data }) => {
+                assert_eq!(data.len(), 1);
+                assert_eq!(data[0].data.len(), 2);
+            }
+            Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+            Err(err) => { panic!("Receive returned error {:?}", err); }
+        }
+        // Let the client test our send_pdata() fragmentation
+        {
+            // Send two more bytes than fit in a PDU
+            let filler_len = max_client_pdu_len - PDV_HDR_LEN + 2;
+            let buf = vec![0_u8; filler_len];
+            let mut sender = association.send_pdata(1);
+            // This should split the data in two packets
+            sender.write_all(&buf).await.expect("Error sending fragmented data");
+        }
+
         // handle one release request
         let pdu = association.receive().await?;
         assert_eq!(pdu, Pdu::ReleaseRQ);
@@ -264,6 +326,35 @@ fn run_scu_scp_association_test(max_is_client: bool) {
         e => panic!("Expected SendTooLongPdu but didn't happen: {:?}", e),
     }
 
+    // Let the server test our send_pdata() fragmentation for us
+    {
+        // Send two more bytes than fit in a PDU
+        let filler_len = max_server_pdu_len - PDV_HDR_LEN + 2;
+        let buf = vec![0_u8; filler_len];
+        let mut sender = association.send_pdata(1);
+        // This should split the data in two packets
+        sender.write_all(&buf).expect("Error sending fragmented data");
+    }
+    // Test send_pdata() fragmentation of the server; we should receive two packets
+    // First packet
+    match association.receive() {
+        Ok(Pdu::PData { data }) => {
+            assert_eq!(data.len(), 1);
+            assert_eq!(data[0].data.len(), max_client_pdu_len - PDV_HDR_LEN);
+        }
+        Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+        Err(err) => { panic!("Receive returned error {:?}", err); }
+    }
+    // Second packet
+    match association.receive() {
+        Ok(Pdu::PData { data }) => {
+            assert_eq!(data.len(), 1);
+            assert_eq!(data[0].data.len(), 2);
+        }
+        Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+        Err(err) => { panic!("Receive returned error {:?}", err); }
+    }
+
     association
         .release()
         .expect("did not have a peaceful release");
@@ -345,6 +436,35 @@ async fn run_scu_scp_association_test_async(max_is_client: bool) {
         e => panic!("Expected SendTooLongPdu but didn't happen (async): {:?}", e),
     }
 
+    // Let the server test our send_pdata() fragmentation for us
+    {
+        // Send two more bytes than fit in a PDU
+        let filler_len = max_server_pdu_len - PDV_HDR_LEN + 2;
+        let buf = vec![0_u8; filler_len];
+        let mut sender = association.send_pdata(1);
+        // This should split the data in two packets
+        sender.write_all(&buf).await.expect("Error sending fragmented data");
+    }
+    // Test send_pdata() fragmentation of the server; we should receive two packets
+    // First packet
+    match association.receive().await {
+        Ok(Pdu::PData { data }) => {
+            assert_eq!(data.len(), 1);
+            assert_eq!(data[0].data.len(), max_client_pdu_len - PDV_HDR_LEN);
+        }
+        Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+        Err(err) => { panic!("Receive returned error {:?}", err); }
+    }
+    // Second packet
+    match association.receive().await {
+        Ok(Pdu::PData { data }) => {
+            assert_eq!(data.len(), 1);
+            assert_eq!(data[0].data.len(), 2);
+        }
+        Ok(other_pdus) => { panic!("Unknown PDU: {:?}", other_pdus); }
+        Err(err) => { panic!("Receive returned error {:?}", err); }
+    }
+
     association
         .release()
         .await