dicom-ul 0.10.0

//! Association acceptor module
//!
//! The module provides an abstraction for a DICOM association
//! in which this application entity listens to incoming association requests.
//! See [`ServerAssociationOptions`]
//! for details and examples on how to create an association.
use bytes::BytesMut;
use std::borrow::Cow;
#[cfg(feature = "sync-tls")]
use std::sync::Arc;
use std::time::Duration;
use std::{io::Write, net::TcpStream};

use crate::association::private::SyncAssociationSealed;
use crate::association::{
    AbortedSnafu, Association, CloseSocket, MissingAbstractSyntaxSnafu, RejectedSnafu,
    SendPduSnafu, SocketOptions, SyncAssociation, UnexpectedPduSnafu, UnknownPduSnafu,
    WireSendSnafu, encode_pdu, read_pdu_from_wire,
};
use dicom_encoding::transfer_syntax::TransferSyntaxIndex;
use dicom_transfer_syntax_registry::TransferSyntaxRegistry;
use snafu::{ResultExt, ensure};

use crate::association::{NegotiatedOptions, RequestorRoles};
use crate::pdu::{LARGE_PDU_SIZE, MAXIMUM_PDU_SIZE, PresentationContextNegotiated};
use crate::{
    IMPLEMENTATION_CLASS_UID, IMPLEMENTATION_VERSION_NAME,
    pdu::{
        AbortRQServiceProviderReason, AbortRQSource, AssociationAC, AssociationRJ,
        AssociationRJResult, AssociationRJServiceUserReason, AssociationRJSource, AssociationRQ,
        DEFAULT_MAX_PDU, PDU_HEADER_SIZE, Pdu, PresentationContextResult,
        PresentationContextResultReason, UserIdentity, UserVariableItem, write_pdu,
    },
};
#[cfg(feature = "sync-tls")]
use tracing::{error, warn};

use super::{Error, Result, uid::trim_uid};

#[cfg(feature = "async")]
use crate::association::AsyncAssociation;

// stray module from 0.9.0, remove in 0.10.0
#[deprecated(since = "0.9.1")]
pub mod non_blocking {}

#[cfg(feature = "sync-tls")]
pub type TlsStream = rustls::StreamOwned<rustls::ServerConnection, std::net::TcpStream>;
#[cfg(feature = "async-tls")]
pub type AsyncTlsStream = tokio_rustls::server::TlsStream<tokio::net::TcpStream>;

#[cfg(feature = "sync-tls")]
use rustls::{
    ServerConfig, ServerConnection,
    server::{Accepted, Acceptor},
};

/// Common interface for application entity access control policies.
///
/// Existing implementations include [`AcceptAny`] and [`AcceptCalledAeTitle`],
/// but users are free to implement their own.
pub trait AccessControl {
    /// Obtain the decision of whether to accept an incoming association request
    /// based on the recorded application entity titles and/or user identity.
    ///
    /// Returns Ok(()) if the requester node should be given clearance.
    /// Otherwise, a concrete association RJ service user reason is given.
    fn check_access(
        &self,
        this_ae_title: &str,
        calling_ae_title: &str,
        called_ae_title: &str,
        user_identity: Option<&UserIdentity>,
    ) -> Result<(), AssociationRJServiceUserReason>;
}

/// An access control rule that accepts any incoming association request.
#[derive(Debug, Default, Copy, Clone, Eq, Hash, PartialEq)]
pub struct AcceptAny;

impl AccessControl for AcceptAny {
    fn check_access(
        &self,
        _this_ae_title: &str,
        _calling_ae_title: &str,
        _called_ae_title: &str,
        _user_identity: Option<&UserIdentity>,
    ) -> Result<(), AssociationRJServiceUserReason> {
        Ok(())
    }
}

/// An access control rule that accepts association requests
/// that match the called AE title with the node's AE title.
#[derive(Debug, Default, Copy, Clone, Eq, Hash, PartialEq)]
pub struct AcceptCalledAeTitle;

impl AccessControl for AcceptCalledAeTitle {
    fn check_access(
        &self,
        this_ae_title: &str,
        _calling_ae_title: &str,
        called_ae_title: &str,
        _user_identity: Option<&UserIdentity>,
    ) -> Result<(), AssociationRJServiceUserReason> {
        if this_ae_title == called_ae_title {
            Ok(())
        } else {
            Err(AssociationRJServiceUserReason::CalledAETitleNotRecognized)
        }
    }
}

/// Interface for negotiation of certain aspects of the association.
pub trait Negotiation {
    /// User-provided extended negotiation. The result of
    /// the negotiation must be returned as either None, if
    /// extended negotiation is not supported by the
    /// application for the given SOP Class, or `Some(Vec<u8>)`
    /// with the result of the negotiation, if the SOP Class is
    /// supported and negotiation is successful. The meaning
    /// is SOP Class-specific and typically index-specific. For
    /// example, for the SOP Class 1.2.840.10008.5.1.4.1.2.2.1
    /// (Study Root Query/Retrieve Information Model - FIND),
    /// the first byte in the output should be 1 if the first
    /// input byte is 1 (client requests relational queries)
    /// and the server supports relational queries, and 0
    /// otherwise, and so on, according to
    /// [Table C.5-2 in PS3.4 of the Standard][1]. That link
    /// applies to the SOP class of this example; see the
    /// appropriate section of the standard for each SOP class
    /// you intend to support.
    ///
    /// [1]: https://dicom.nema.org/medical/dicom/2025d/output/chtml/part04/sect_C.5.html#table_C.5-2
    ///
    /// The default action is to not perform extended negotiation
    /// (returns `None` for all SOP Classes, therefore no Extended
    /// Negotiation items are included in the A-ASSOCIATE-AC
    /// response).
    ///
    /// When implementing an extended_negotiation method, never
    /// return more bytes than the version of the standard you
    /// support specifies. If you're using the input array as a
    /// basis, truncate or slice it accordingly before returning
    /// it.
    ///
    /// This is an example of a server that supports combined
    /// date/time matching and empty value matching for the
    /// SOP Class "Study Root Query/Retrieve Information Model - FIND":
    /// ```no_run
    /// # use dicom_ul::association::server::{Negotiation, ServerAssociationOptions};
    /// # use dicom_ul::association::Association;
    /// # use std::net::TcpListener;
    /// const STUDY_ROOT_QR_IM_FIND: &str = "1.2.840.10008.5.1.4.1.2.2.1";
    ///
    /// struct FindOptions;
    ///
    /// impl Negotiation for FindOptions {
    ///     fn extended_negotiation(
    ///         &self,
    ///         sop_class_uid: &str,
    ///         input: &[u8],
    ///     ) -> Option<Vec<u8>> {
    ///         // Only this SOP Class is supported, so ditch the rest
    ///         if sop_class_uid != STUDY_ROOT_QR_IM_FIND {
    ///             return None;
    ///         }
    ///         // Truncate to at most 7 elements (those supported by the 2025d
    ///         // version of the standard) and store in a new vector.
    ///         let mut result = input[..input.len().min(7)].to_vec();
    ///         for index in 1..=result.len() {
    ///             match index {
    ///                 // Combined date-time matching (second byte of field) and
    ///                 // Empty value matching (sixth byte) are both supported.
    ///                 // Remember that the DICOM standard specifies 1-based
    ///                 // byte positions, but Rust indices are 0-based.
    ///                 2 | 6 => (),
    ///                 // Indicate lack of support for the rest of features.
    ///                 _ => { result[index - 1] = 0; }
    ///             }
    ///         }
    ///         Some(result)
    ///     }
    /// }
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let tcp_listener: TcpListener = unimplemented!();
    /// let find_options = FindOptions;
    ///
    /// let scp_options = ServerAssociationOptions::new()
    ///     .with_negotiation(find_options);
    ///
    /// let (stream, _address) = tcp_listener.accept()?;
    /// let assoc = scp_options.establish(stream)?;
    ///
    /// // Association is established; read the flags that were negotiated
    /// let mut combined_date_time = false;
    /// let mut empty_value_matching = false;
    ///
    /// if let Some(values) = assoc.extended_negotiation_for(STUDY_ROOT_QR_IM_FIND) {
    ///     // 2nd byte position (index 1) is the combined date/time matching option.
    ///     // If it is zero or absent, consider it disabled.
    ///     combined_date_time = values.get(2 - 1).copied().unwrap_or(0) != 0;
    ///     // 6th byte position (index 5) is the empty value matching option
    ///     empty_value_matching = values.get(6 - 1).copied().unwrap_or(0) != 0;
    /// }
    /// # Ok(())
    /// # }
    /// ```
    #[allow(unused_variables)]
    fn extended_negotiation(&self, sop_class_uid: &str, input: &[u8]) -> Option<Vec<u8>> {
        None
    }

    /// User-provided negotiation of accepted association-requestor
    /// (client) roles for a given SOP Class. The client may request
    /// to be either an SCU, an SCP, or both for any given SOP Class.
    /// Through this function, the server will decide for that SOP
    /// Class whether to allow the client to take that role or not.
    /// Note that if the client specifies `false` for one of its
    /// roles, it's not allowed for the server to return `true` for
    /// that role in that SOP Class.
    ///
    /// The return value is an `Option<(bool, bool)>`. It shall be
    /// `None` if this negotiation item is not to be returned to the
    /// client, or `Some((scu_role, scp_role))` with the accepted
    /// values.
    ///
    /// The default action is to ignore this negotiation and not
    /// send back any role selection for any SOP CLass, which
    /// implicitly makes the client be an SCU.
    ///
    /// In this example, the server rejects any client's proposal
    /// to be an SCP for the Storage MR SOP Class:
    /// ```no_run
    /// # use dicom_ul::association::server::{Negotiation, ServerAssociationOptions};
    /// # use dicom_ul::association::Association;
    /// # use dicom_ul::pdu::RequestorRoles;
    /// # use std::net::TcpListener;
    /// const STORAGE_MR_SOP_CLASS: &str = "1.2.840.10008.5.1.4.1.1.4";
    /// struct RoleOptions;
    ///
    /// impl Negotiation for RoleOptions {
    ///     fn negotiate_roles(
    ///         &self,
    ///         sop_class_uid: &str,
    ///         scu_role: bool,
    ///         scp_role: bool,
    ///     ) -> Option<RequestorRoles> {
    ///         // Only this SOP Class is supported, so ditch the rest
    ///         if sop_class_uid != STORAGE_MR_SOP_CLASS {
    ///             return None;
    ///         }
    ///         // Always set SCP role to false
    ///         Some(RequestorRoles { scu: scu_role, scp: false })
    ///     }
    /// }
    ///
    /// # fn main() -> Result<(), Box<dyn std::error::Error>> {
    /// # let tcp_listener: TcpListener = unimplemented!();
    /// let role_options = RoleOptions;
    ///
    /// let scp_options = ServerAssociationOptions::new()
    ///     .with_negotiation(role_options);
    ///
    /// let (stream, _address) = tcp_listener.accept()?;
    /// let assoc = scp_options.establish(stream)?;
    ///
    /// // Association is established; read the flags that were negotiated
    /// // (or not) for the association requestor.
    /// let RequestorRoles { scu: client_is_scu, scp: client_is_scp } =
    ///     assoc.requestor_roles_for(STORAGE_MR_SOP_CLASS);
    /// # Ok(())
    /// # }
    /// ```
    #[allow(unused_variables)]
    fn negotiate_roles(
        &self,
        sop_class_uid: &str,
        scu_role: bool,
        scp_role: bool,
    ) -> Option<RequestorRoles> {
        None
    }
}

/// A Negotiation rule that causes the server to omit the negotiation in the response, thus forcing
/// default values for all kinds of negotiations supported, for all SOP classes.
#[derive(Debug, Default, Copy, Clone, Eq, Hash, PartialEq)]
pub struct DefaultNegotiation;

impl Negotiation for DefaultNegotiation {}

#[cfg(feature = "sync-tls")]
/// Attempt to accept an incoming TLS connection
///
/// Returns
/// * `Ok(Ok(Accepted))` when the connection is accepted
/// * `Ok(Err(err, alert))` when the connection was not accepted against bare config
/// * `Err(TlsHandshakeSnafu)` otherwise (on io error)
fn accept_tls<S>(
    socket: &mut S,
) -> Result<std::result::Result<Accepted, (rustls::Error, rustls::server::AcceptedAlert)>>
where
    S: std::io::Read + std::io::Write,
{
    let mut acceptor = Acceptor::default();
    let accepted = loop {
        acceptor
            .read_tls(socket)
            .context(super::TlsHandshakeSnafu)?;
        match acceptor.accept() {
            Ok(Some(accepted)) => break accepted,
            Ok(None) => {}
            Err((err, alert)) => return Ok(Err((err, alert))),
        }
    };
    Ok(Ok(accepted))
}

#[cfg(feature = "sync-tls")]
/// Attempt to handshake an incoming TLS connection
///
/// First attempts to accept using a bare config, then validates using the
/// provided ServerConfig
///
/// If there is an error in either, a `rustls::server::AcceptedAlert` will be
/// written back to the client
fn handshake_tls(
    socket: &mut TcpStream,
    tls_config: Arc<ServerConfig>,
) -> Result<ServerConnection> {
    let maybe_tls = accept_tls(socket)?;
    match maybe_tls {
        Ok(accepted) => {
            // Connection was accepted, now validate against our ServerConfig and send an alert
            // if validation fails
            Ok(accepted.into_connection(tls_config.clone()).map_err(|(err, mut alert)| {
                let peer = socket.peer_addr().map(|addr| addr.to_string()).unwrap_or("unknown".to_string());
                warn!(remote = peer, err = ?err, "Incoming TLS connection not accepted, sending alert to client.");
                if let Err(e) = alert.write_all(socket){
                    error!("Failed to send alert notification to client: {:?}", e)
                }
                err
            }).context(super::TlsConnectionSnafu)?)
        }
        Err((err, mut alert)) => {
            let peer = socket
                .peer_addr()
                .map(|addr| addr.to_string())
                .unwrap_or("unknown".to_string());
            warn!(remote = peer, err = ?err, "Incoming TLS connection not accepted, sending alert to client.");
            // Connection was not accepted, send an alert as to why
            if let Err(e) = alert.write_all(socket) {
                error!("Failed to send alert notification to client: {:?}", e)
            }
            Err(err).context(super::TlsConnectionSnafu)
        }
    }
}

/// A DICOM association builder for an acceptor DICOM node,
/// often taking the role of a service class provider (SCP).
///
/// This is the standard way of negotiating and establishing
/// an association with a requesting node.
/// The outcome is a [`ServerAssociation`].
/// Unlike the [`ClientAssociationOptions`],
/// a value of this type can be reused for multiple connections.
///
/// [`ClientAssociationOptions`]: crate::association::ClientAssociationOptions
///
/// The SCP will by default accept all transfer syntaxes
/// supported by the main [transfer syntax registry][1],
/// unless one or more transfer syntaxes are explicitly indicated
/// through calls to [`with_transfer_syntax`][2].
///
/// Access control logic is also available,
/// enabling application entities to decide on
/// whether to accept or reject the association request
/// based on the _called_ and _calling_ AE titles.
///
/// - By default, the application will accept requests from anyone
///   ([`AcceptAny`])
/// - To only accept requests with a matching _called_ AE title,
///   add a call to [`accept_called_ae_title`]
///   ([`AcceptCalledAeTitle`]).
/// - Any other policy can be implemented through the [`AccessControl`] trait.
///
/// [`accept_called_ae_title`]: Self::accept_called_ae_title
/// [`AcceptAny`]: AcceptAny
/// [`AcceptCalledAeTitle`]: AcceptCalledAeTitle
/// [`AccessControl`]: AccessControl
///
/// [1]: dicom_transfer_syntax_registry
/// [2]: ServerAssociationOptions::with_transfer_syntax
///
/// ## Basic Usage
///
/// ### Synchronous API
///
/// Spawn a single sync thread to listen for incoming requests.
/// ```no_run
/// # use std::net::TcpListener;
/// # use dicom_ul::association::server::ServerAssociationOptions;
/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
/// # let tcp_listener: TcpListener = unimplemented!();
/// let scp_options = ServerAssociationOptions::new()
///    .with_abstract_syntax("1.2.840.10008.1.1")
///    .with_transfer_syntax("1.2.840.10008.1.2.1");
///
/// let (stream, _address) = tcp_listener.accept()?;
/// scp_options.establish(stream)?;
/// # Ok(())
/// # }
/// ```
///
/// ### Asynchronous API
///
/// Spawn an async task for each incoming association request.
///
/// ```no_run
/// # use std::net::{Ipv4Addr, SocketAddrV4};
/// # use dicom_ul::association::{server::ServerAssociationOptions};
/// # #[cfg(feature = "async")]
/// # #[tokio::main]
/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
/// # use dicom_ul::association::AsyncAssociation;
/// let listen_addr = SocketAddrV4::new(Ipv4Addr::from(0), 11111);
/// let listener = tokio::net::TcpListener::bind(listen_addr).await?;
/// loop {
///     let (socket, _addr) = listener.accept().await?;
///     tokio::task::spawn(async move {
///         let mut scp = ServerAssociationOptions::new()
///             .accept_any()
///             .with_abstract_syntax("1.2.840.10008.1.1")
///             .with_transfer_syntax("1.2.840.10008.1.2.1")
///             .establish_async(socket)
///             .await
///             .expect("Could not establish association on socket");
///         loop {
///             match scp.receive().await {
///                 Ok(dicom_ul::Pdu::PData { data }) => {
///                     // read P-Data here
///                 },
///                 Ok(dicom_ul::Pdu::ReleaseRP) => {
///                     break;
///                 },
///                 Ok(dicom_ul::Pdu::AbortRQ { source }) => {
///                     eprintln!("Association aborted: {source:?}");
///                     break;
///                 },
///                 Ok(pdu) => {
///                     eprintln!("Unexpected PDU");
///                 },
///                 Err(e) => {
///                     eprintln!("Oops! {e}");
///                 },
///             }
///         }
///     });
/// }
/// # Ok(())
/// # }
/// # #[cfg(not(feature = "async"))]
/// fn main() {}
/// ```
///
/// ## TLS Support
///
/// Enabling one of the Cargo features `sync-tls` or `async-tls`
/// unlocks the methods for configuring TLS.
/// Call `tls_config`
/// for the server to expect associations established
/// over a secure transport connection.
///
/// #### TLS in synchronous API
///
/// Include the `sync-tls` feature in your `Cargo.toml`.
///
/// #### TLS in asynchronous API
///
/// Include the `async-tls` feature in your `Cargo.toml`.
///
/// ### Example
///
/// ```no_run
/// # use std::time::Duration;
/// # use std::sync::Arc;
/// # #[cfg(feature = "sync-tls")]
/// # fn run() -> Result<(), Box<dyn std::error::Error>> {
/// use dicom_ul::{ServerAssociation, ServerAssociationOptions};
/// use std::net::TcpListener;
/// use rustls::{
///     ServerConfig, RootCertStore,
///     pki_types::{CertificateDer, PrivateKeyDer, pem::PemObject},
///     server::WebPkiClientVerifier,
/// };
/// # let tcp_listener: TcpListener = unimplemented!();
/// // Loading certificates and keys for demonstration purposes
/// let ca_cert = CertificateDer::from_pem_slice(std::fs::read("ssl/ca.crt")?.as_ref())
///     .expect("Failed to load client cert");
///
/// // Server certificate and private key -- signed by CA
/// let server_cert = CertificateDer::from_pem_slice(std::fs::read("ssl/server.crt")?.as_ref())
///     .expect("Failed to load server cert");
///
/// let server_private_key = PrivateKeyDer::from_pem_slice(std::fs::read("ssl/server.key")?.as_ref())
///     .expect("Failed to load client private key");
///
/// // Create a root cert store for the client which includes the server certificate
/// let mut certs = RootCertStore::empty();
/// certs.add_parsable_certificates(vec![ca_cert.clone()]);
///
/// // Server configuration.
/// // Creates a server config that requires client authentication (mutual TLS) using
/// // webpki for certificate verification.
/// let server_config = ServerConfig::builder()
///     .with_client_cert_verifier(
///         WebPkiClientVerifier::builder(certs.clone().into())
///             .build()
///             .expect("Failed to create client certificate verifier")
///     )
///     .with_single_cert(vec![server_cert.clone(), ca_cert.clone()], server_private_key)
///     .expect("Failed to create server TLS config");
///
/// let (stream, _address) = tcp_listener.accept()?;
///
/// let association: ServerAssociation<_> = ServerAssociationOptions::new()
///     .accept_called_ae_title()
///     .ae_title("TLS-SCP")
///     .with_abstract_syntax(dicom_dictionary_std::uids::VERIFICATION)
///     .tls_config(server_config)
///     .establish_tls(stream)?;
/// # Ok(())
/// # }
/// ```
///
/// For an association with the async API,
/// call `establish_tls_async` instead of `establish_tls`.
#[derive(Debug, Clone)]
pub struct ServerAssociationOptions<'a, A, N> {
    /// the application entity access control policy
    ae_access_control: A,
    /// the AE title of this DICOM node
    ae_title: Cow<'a, str>,
    /// the requested application context name
    application_context_name: Cow<'a, str>,
    /// the list of requested abstract syntaxes
    abstract_syntax_uids: Vec<Cow<'a, str>>,
    /// the list of requested transfer syntaxes
    transfer_syntax_uids: Vec<Cow<'a, str>>,
    /// the expected protocol version
    protocol_version: u16,
    /// the maximum PDU length
    max_pdu_length: u32,
    /// whether to receive PDUs in strict mode
    strict: bool,
    /// whether to accept unknown abstract syntaxes
    promiscuous: bool,
    /// extended negotiation handler
    negotiation: N,
    /// Options for the underlying TCP socket
    socket_options: SocketOptions,
    /// TLS configuration for the underlying TCP socket
    #[cfg(feature = "sync-tls")]
    tls_config: Option<std::sync::Arc<rustls::ServerConfig>>,
}

impl Default for ServerAssociationOptions<'_, AcceptAny, DefaultNegotiation> {
    fn default() -> Self {
        ServerAssociationOptions {
            ae_access_control: AcceptAny,
            ae_title: "THIS-SCP".into(),
            application_context_name: "1.2.840.10008.3.1.1.1".into(),
            abstract_syntax_uids: Vec::new(),
            transfer_syntax_uids: Vec::new(),
            protocol_version: 1,
            max_pdu_length: DEFAULT_MAX_PDU,
            strict: true,
            promiscuous: false,
            negotiation: DefaultNegotiation,
            socket_options: SocketOptions::default(),
            #[cfg(feature = "sync-tls")]
            tls_config: None,
        }
    }
}

impl ServerAssociationOptions<'_, AcceptAny, DefaultNegotiation> {
    /// Create a new set of options for establishing an association.
    pub fn new() -> Self {
        Self::default()
    }
}

impl<'a, A, N> ServerAssociationOptions<'a, A, N>
where
    A: AccessControl,
    N: Negotiation,
{
    /// Change the access control policy to accept any association
    /// regardless of the specified AE titles.
    ///
    /// This is the default behavior when the options are first created.
    pub fn accept_any(self) -> ServerAssociationOptions<'a, AcceptAny, N> {
        self.ae_access_control(AcceptAny)
    }

    /// Change the access control policy to accept an association
    /// if the called AE title matches this node's AE title.
    ///
    /// The default is to accept any requesting node
    /// regardless of the specified AE titles.
    pub fn accept_called_ae_title(self) -> ServerAssociationOptions<'a, AcceptCalledAeTitle, N> {
        self.ae_access_control(AcceptCalledAeTitle)
    }

    /// Change the access control policy.
    ///
    /// The default is to accept any requesting node
    /// regardless of the specified AE titles.
    pub fn ae_access_control<P>(self, access_control: P) -> ServerAssociationOptions<'a, P, N>
    where
        P: AccessControl,
    {
        let ServerAssociationOptions {
            ae_title,
            application_context_name,
            abstract_syntax_uids,
            transfer_syntax_uids,
            protocol_version,
            max_pdu_length,
            strict,
            promiscuous,
            ae_access_control: _,
            negotiation,
            socket_options,
            #[cfg(feature = "sync-tls")]
            tls_config,
        } = self;

        ServerAssociationOptions {
            ae_access_control: access_control,
            ae_title,
            application_context_name,
            abstract_syntax_uids,
            transfer_syntax_uids,
            protocol_version,
            max_pdu_length,
            strict,
            promiscuous,
            negotiation,
            socket_options,
            #[cfg(feature = "sync-tls")]
            tls_config,
        }
    }

    /// Define the application entity title referring to this DICOM node.
    ///
    /// The default is `THIS-SCP`.
    pub fn ae_title<T>(mut self, ae_title: T) -> Self
    where
        T: Into<Cow<'a, str>>,
    {
        self.ae_title = ae_title.into();
        self
    }

    /// Include this abstract syntax
    /// in the list of proposed presentation contexts.
    pub fn with_abstract_syntax<T>(mut self, abstract_syntax_uid: T) -> Self
    where
        T: Into<Cow<'a, str>>,
    {
        self.abstract_syntax_uids
            .push(trim_uid(abstract_syntax_uid.into()));
        self
    }

    /// Include this transfer syntax in each proposed presentation context.
    pub fn with_transfer_syntax<T>(mut self, transfer_syntax_uid: T) -> Self
    where
        T: Into<Cow<'a, str>>,
    {
        self.transfer_syntax_uids
            .push(trim_uid(transfer_syntax_uid.into()));
        self
    }

    /// Override the maximum expected PDU length.
    /// Values greater than MAXIMUM_PDU_SIZE will be
    /// silently truncated to MAXIMUM_PDU_SIZE.
    pub fn max_pdu_length(mut self, value: u32) -> Self {
        self.max_pdu_length = value.min(MAXIMUM_PDU_SIZE);
        self
    }

    /// Override strict mode:
    /// whether receiving PDUs must not
    /// surpass the negotiated maximum PDU length.
    pub fn strict(mut self, strict: bool) -> Self {
        self.strict = strict;
        self
    }

    /// Override promiscuous mode:
    /// whether to accept unknown abstract syntaxes.
    pub fn promiscuous(mut self, promiscuous: bool) -> Self {
        self.promiscuous = promiscuous;
        self
    }

    /// Set the read timeout for the underlying TCP socket
    ///
    /// This is used to set both the read and write timeout.
    pub fn read_timeout(self, timeout: Duration) -> Self {
        Self {
            socket_options: SocketOptions {
                read_timeout: Some(timeout),
                write_timeout: self.socket_options.write_timeout,
                connection_timeout: self.socket_options.connection_timeout,
            },
            ..self
        }
    }

    /// Set the write timeout for the underlying TCP socket
    pub fn write_timeout(self, timeout: Duration) -> Self {
        Self {
            socket_options: SocketOptions {
                read_timeout: self.socket_options.read_timeout,
                write_timeout: Some(timeout),
                connection_timeout: self.socket_options.connection_timeout,
            },
            ..self
        }
    }

    /// Set the extended negotiation handler
    pub fn with_negotiation<NN>(self, negotiation: NN) -> ServerAssociationOptions<'a, A, NN>
    where
        NN: Negotiation,
    {
        let ServerAssociationOptions {
            ae_access_control,
            ae_title,
            application_context_name,
            abstract_syntax_uids,
            transfer_syntax_uids,
            protocol_version,
            max_pdu_length,
            strict,
            promiscuous,
            negotiation: _,
            socket_options,
            #[cfg(feature = "sync-tls")]
            tls_config,
        } = self;

        ServerAssociationOptions {
            ae_access_control,
            ae_title,
            application_context_name,
            abstract_syntax_uids,
            transfer_syntax_uids,
            protocol_version,
            max_pdu_length,
            strict,
            promiscuous,
            negotiation,
            socket_options,
            #[cfg(feature = "sync-tls")]
            tls_config,
        }
    }

    /// Set the TLS configuration for the underlying TCP socket
    #[cfg(feature = "sync-tls")]
    pub fn tls_config(mut self, config: impl Into<std::sync::Arc<rustls::ServerConfig>>) -> Self {
        self.tls_config = Some(config.into());
        self
    }

    /// Process an association request PDU
    ///
    /// In the success case, returns
    /// * Pdu to be written back to client
    /// * Negotiated options
    /// * Calling AE title
    ///
    /// In the error case, returns
    /// * Pdu to be written back to client
    /// * Error
    #[allow(clippy::result_large_err)]
    fn process_a_association_rq(
        &self,
        msg: Pdu,
    ) -> std::result::Result<(Pdu, NegotiatedOptions, String), (Pdu, Error)> {
        match msg {
            Pdu::AssociationRQ(AssociationRQ {
                protocol_version,
                calling_ae_title,
                called_ae_title,
                application_context_name,
                presentation_contexts,
                user_variables,
            }) => {
                if protocol_version != self.protocol_version {
                    let association_rj = AssociationRJ {
                        result: AssociationRJResult::Permanent,
                        source: AssociationRJSource::ServiceUser(
                            AssociationRJServiceUserReason::NoReasonGiven,
                        ),
                    };
                    let pdu = Pdu::AssociationRJ(association_rj.clone());
                    return Err((pdu, RejectedSnafu { association_rj }.build()));
                }

                if application_context_name != self.application_context_name {
                    let association_rj = AssociationRJ {
                        result: AssociationRJResult::Permanent,
                        source: AssociationRJSource::ServiceUser(
                            AssociationRJServiceUserReason::ApplicationContextNameNotSupported,
                        ),
                    };
                    let pdu = Pdu::AssociationRJ(association_rj.clone());
                    return Err((pdu, RejectedSnafu { association_rj }.build()));
                }

                // User variables resulting from the negotiation are stored here
                let mut new_user_variables = vec![
                    UserVariableItem::MaxLength(self.max_pdu_length),
                    UserVariableItem::ImplementationClassUID(IMPLEMENTATION_CLASS_UID.to_string()),
                    UserVariableItem::ImplementationVersionName(
                        IMPLEMENTATION_VERSION_NAME.to_string(),
                    ),
                ];

                // Process user variables in a single pass
                let mut user_identity = None;
                let mut requestor_max_pdu_length = DEFAULT_MAX_PDU;
                for user_variable in user_variables {
                    use UserVariableItem::*;
                    match user_variable {
                        UserIdentityItem(negotiated_user_identity) => {
                            user_identity = Some(negotiated_user_identity);
                        }

                        MaxLength(len) => {
                            requestor_max_pdu_length = if len == 0 {
                                // treat 0 as the largest we support
                                MAXIMUM_PDU_SIZE
                            } else {
                                // Don't accept more than MAXIMUM_PDU_SIZE
                                // as max_pdu_length.
                                len.min(MAXIMUM_PDU_SIZE)
                            };
                        }

                        // Extended negotiation is performed here: add only the variables
                        // for which the user-supplied negotiation function returns Some(x).
                        SopClassExtendedNegotiationSubItem(sop_class_uid, scai) => {
                            if let Some(extended_negotiation_result) =
                                self.negotiation.extended_negotiation(&sop_class_uid, &scai)
                            {
                                new_user_variables.push(SopClassExtendedNegotiationSubItem(
                                    sop_class_uid.clone(),
                                    extended_negotiation_result,
                                ));
                            }
                        }

                        // SCU/SCP Role Selection negotiation is performed here: add only the
                        // variables for which the user-supplied negotiation function returns
                        // Some(x).
                        ScuScpRoleSelectionSubItem(
                            sop_class_uid,
                            RequestorRoles {
                                scu: scu_requested,
                                scp: scp_requested,
                            },
                        ) => {
                            if let Some(RequestorRoles {
                                scu: scu_accepted,
                                scp: scp_accepted,
                            }) = self.negotiation.negotiate_roles(
                                &sop_class_uid,
                                scu_requested,
                                scp_requested,
                            ) {
                                new_user_variables.push(ScuScpRoleSelectionSubItem(
                                    sop_class_uid,
                                    RequestorRoles {
                                        scu: scu_accepted,
                                        scp: scp_accepted,
                                    },
                                ));
                            }
                        }

                        _ => (),
                    }
                }

                self.ae_access_control
                    .check_access(
                        &self.ae_title,
                        &calling_ae_title,
                        &called_ae_title,
                        user_identity.as_ref(),
                    )
                    .map(Ok)
                    .unwrap_or_else(|reason| {
                        let association_rj = AssociationRJ {
                            result: AssociationRJResult::Permanent,
                            source: AssociationRJSource::ServiceUser(reason),
                        };
                        let pdu = Pdu::AssociationRJ(association_rj.clone());
                        Err((pdu, RejectedSnafu { association_rj }.build()))
                    })?;

                let presentation_contexts_negotiated: Vec<_> = presentation_contexts
                    .into_iter()
                    .map(|pc| {
                        let abstract_syntax = trim_uid(Cow::from(pc.abstract_syntax));
                        if !self.abstract_syntax_uids.contains(&abstract_syntax)
                            && !self.promiscuous
                        {
                            return PresentationContextNegotiated {
                                id: pc.id,
                                reason: PresentationContextResultReason::AbstractSyntaxNotSupported,
                                transfer_syntax: "1.2.840.10008.1.2".to_string(),
                                abstract_syntax: abstract_syntax.to_string(),
                            };
                        }

                        let (transfer_syntax, reason) = self
                            .choose_ts(pc.transfer_syntaxes)
                            .map(|ts| (ts, PresentationContextResultReason::Acceptance))
                            .unwrap_or_else(|| {
                                (
                                    "1.2.840.10008.1.2".to_string(),
                                    PresentationContextResultReason::TransferSyntaxesNotSupported,
                                )
                            });

                        PresentationContextNegotiated {
                            id: pc.id,
                            reason,
                            transfer_syntax,
                            abstract_syntax: abstract_syntax.to_string(),
                        }
                    })
                    .collect();

                let pdu = Pdu::AssociationAC(AssociationAC {
                    protocol_version: self.protocol_version,
                    application_context_name,
                    presentation_contexts: presentation_contexts_negotiated
                        .iter()
                        .map(|pc| PresentationContextResult {
                            id: pc.id,
                            reason: pc.reason.clone(),
                            transfer_syntax: pc.transfer_syntax.clone(),
                        })
                        .collect(),
                    calling_ae_title: calling_ae_title.clone(),
                    called_ae_title: called_ae_title.clone(),
                    user_variables: new_user_variables.clone(),
                });
                Ok((
                    pdu,
                    NegotiatedOptions {
                        peer_max_pdu_length: requestor_max_pdu_length,
                        user_variables: new_user_variables,
                        presentation_contexts: presentation_contexts_negotiated,
                        peer_ae_title: calling_ae_title,
                    },
                    called_ae_title,
                ))
            }
            Pdu::ReleaseRQ => Err((Pdu::ReleaseRP, AbortedSnafu.build())),
            pdu @ Pdu::AssociationAC { .. }
            | pdu @ Pdu::AssociationRJ { .. }
            | pdu @ Pdu::PData { .. }
            | pdu @ Pdu::ReleaseRP
            | pdu @ Pdu::AbortRQ { .. } => Err((
                Pdu::AbortRQ {
                    source: AbortRQSource::ServiceProvider(
                        AbortRQServiceProviderReason::UnexpectedPdu,
                    ),
                },
                UnexpectedPduSnafu { pdu }.build(),
            )),
            pdu @ Pdu::Unknown { .. } => Err((
                Pdu::AbortRQ {
                    source: AbortRQSource::ServiceProvider(
                        AbortRQServiceProviderReason::UnrecognizedPdu,
                    ),
                },
                UnknownPduSnafu { pdu }.build(),
            )),
        }
    }

    /// Negotiate an association with the given TCP stream.
    pub fn establish(&self, mut socket: TcpStream) -> Result<ServerAssociation<TcpStream>> {
        ensure!(
            !self.abstract_syntax_uids.is_empty() || self.promiscuous,
            MissingAbstractSyntaxSnafu
        );

        socket
            .set_read_timeout(self.socket_options.read_timeout)
            .context(super::SetReadTimeoutSnafu)?;
        socket
            .set_write_timeout(self.socket_options.write_timeout)
            .context(super::SetWriteTimeoutSnafu)?;

        let mut read_buffer = BytesMut::with_capacity(
            (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
        );
        let msg = read_pdu_from_wire(
            &mut socket,
            &mut read_buffer,
            self.max_pdu_length,
            self.strict,
        );
        // If we're compiling with the sync-tls feature, check to see if the error
        // may have been caused by the client associating with TLS but the server
        // not being run with TLS support
        #[cfg(feature = "sync-tls")]
        let msg = msg.map_err(|e| {
            // read_pdu_from_wire consumes bytes on the socket, but puts them into read_buffer
            let mut cursor = std::io::Cursor::new(read_buffer.to_vec());
            if let Ok(Ok(_)) = accept_tls(&mut cursor) {
                // If the connection was accepted, return an error, TLS
                // client attempting to connect with non-TLS server
                return super::TlsNotSupportedSnafu.build();
            }
            e
        });
        let msg = msg?;
        let mut write_buffer: Vec<u8> =
            Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
        match self.process_a_association_rq(msg) {
            Ok((
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            )) => {
                write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                socket.write_all(&write_buffer).context(WireSendSnafu)?;
                Ok(ServerAssociation {
                    presentation_contexts,
                    requestor_max_pdu_length: peer_max_pdu_length,
                    acceptor_max_pdu_length: self.max_pdu_length,
                    socket,
                    client_ae_title: peer_ae_title,
                    write_buffer,
                    strict: self.strict,
                    read_buffer,
                    user_variables,
                    called_ae_title,
                })
            }
            Err((pdu, err)) => {
                // send the rejection/abort PDU
                write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                socket.write_all(&write_buffer).context(WireSendSnafu)?;
                Err(err)
            }
        }
    }

    /// Negotiate an association with the given TCP stream using TLS.
    #[cfg(feature = "sync-tls")]
    pub fn establish_tls(&self, mut socket: TcpStream) -> Result<ServerAssociation<TlsStream>> {
        ensure!(
            !self.abstract_syntax_uids.is_empty() || self.promiscuous,
            MissingAbstractSyntaxSnafu
        );
        let tls_config = self
            .tls_config
            .as_ref()
            .ok_or_else(|| super::TlsConfigMissingSnafu {}.build())?;

        socket
            .set_read_timeout(self.socket_options.read_timeout)
            .context(super::SetReadTimeoutSnafu)?;
        socket
            .set_write_timeout(self.socket_options.write_timeout)
            .context(super::SetWriteTimeoutSnafu)?;

        let conn = handshake_tls(&mut socket, tls_config.clone())?;
        let mut tls_stream = rustls::StreamOwned::new(conn, socket);
        let mut read_buffer = BytesMut::with_capacity(
            (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
        );

        let msg = read_pdu_from_wire(
            &mut tls_stream,
            &mut read_buffer,
            self.max_pdu_length,
            self.strict,
        )?;
        let mut write_buffer: Vec<u8> =
            Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
        match self.process_a_association_rq(msg) {
            Ok((
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            )) => {
                write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                tls_stream.write_all(&write_buffer).context(WireSendSnafu)?;
                Ok(ServerAssociation {
                    presentation_contexts,
                    requestor_max_pdu_length: peer_max_pdu_length,
                    acceptor_max_pdu_length: self.max_pdu_length,
                    socket: tls_stream,
                    client_ae_title: peer_ae_title,
                    write_buffer,
                    strict: self.strict,
                    read_buffer,
                    user_variables,
                    called_ae_title,
                })
            }
            Err((pdu, err)) => {
                // send the rejection/abort PDU
                write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                tls_stream.write_all(&write_buffer).context(WireSendSnafu)?;
                Err(err)
            }
        }
    }

    /// From a sequence of transfer syntaxes,
    /// choose the first transfer syntax to
    /// - be on the options' list of transfer syntaxes, and
    /// - be supported by the main transfer syntax registry.
    ///
    /// If the options' list is empty,
    /// accept the first transfer syntax supported.
    fn choose_ts<I, T>(&self, it: I) -> Option<T>
    where
        I: IntoIterator<Item = T>,
        T: AsRef<str>,
    {
        if self.transfer_syntax_uids.is_empty() {
            return choose_supported(it);
        }

        it.into_iter().find(|ts| {
            let ts = ts.as_ref();
            if self.transfer_syntax_uids.is_empty() {
                ts.trim_end_matches(|c: char| c.is_whitespace() || c == '\0') == "1.2.840.10008.1.2"
            } else {
                self.transfer_syntax_uids.contains(&trim_uid(ts.into())) && is_supported(ts)
            }
        })
    }
}

/// A DICOM upper level association from the perspective
/// of an accepting application entity.
///
/// The most common operations of an established association are
/// [`send`](SyncAssociation::send)
/// and [`receive`](SyncAssociation::receive).
/// Sending large P-Data fragments may be easier through the P-Data sender
/// abstraction (see [`send_pdata`](SyncAssociation::send_pdata)).
///
/// When the value falls out of scope,
/// the program will shut down the underlying TCP connection.
#[derive(Debug)]
pub struct ServerAssociation<S> {
    /// The accorded presentation contexts
    presentation_contexts: Vec<PresentationContextNegotiated>,
    /// The maximum PDU length that the remote application entity accepts
    requestor_max_pdu_length: u32,
    /// The maximum PDU length that this application entity is expecting to receive
    acceptor_max_pdu_length: u32,
    /// The TCP stream to the other DICOM node
    socket: S,
    /// The application entity title of the other DICOM node
    client_ae_title: String,
    /// The application entity title used by the client
    /// to call this server
    called_ae_title: String,
    /// Reusable buffer used for sending PDUs on the wire
    /// prevents reallocation on each send
    write_buffer: Vec<u8>,
    /// whether to receive PDUs in strict mode
    strict: bool,
    /// Read buffer from the socket
    read_buffer: bytes::BytesMut,
    /// User variables received from the peer
    user_variables: Vec<UserVariableItem>,
}

// compatibility filler, remove in 0.10.0
impl<S> ServerAssociation<S> {
    /// Obtain a view of the negotiated presentation contexts.
    pub fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
        &self.presentation_contexts
    }

    /// Retrieve the maximum PDU length
    /// that the association acceptor is expecting to receive.
    pub 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.
    pub fn requestor_max_pdu_length(&self) -> u32 {
        self.requestor_max_pdu_length
    }

    /// Obtain the remote DICOM node's application entity title.
    #[deprecated(
        since = "0.9.1",
        note = "Call `peer_ae_title` from trait `Association`"
    )]
    pub fn client_ae_title(&self) -> &str {
        &self.client_ae_title
    }

    /// Obtain the application entity title called by the client.
    pub fn called_ae_title(&self) -> &str {
        &self.called_ae_title
    }
}

impl<S> ServerAssociation<S>
where
    S: std::io::Read + std::io::Write + CloseSocket,
{
    /// Send a PDU message to the other intervenient.
    pub fn send(&mut self, msg: &Pdu) -> Result<()> {
        SyncAssociation::send(self, msg)
    }

    /// Read a PDU message from the other intervenient.
    pub fn receive(&mut self) -> Result<Pdu> {
        SyncAssociation::receive(self)
    }

    /// Send a provider initiated abort message
    /// and shut down the TCP connection,
    /// terminating the association.
    pub fn abort(self) -> Result<()> {
        SyncAssociation::abort(self)
    }

    /// Prepare a P-Data writer for sending
    /// one or more data item PDUs.
    ///
    /// Returns a writer which automatically
    /// splits the inner data into separate PDUs if necessary.
    pub fn send_pdata(
        &mut self,
        presentation_context_id: u8,
    ) -> crate::association::pdata::PDataWriter<&mut S> {
        SyncAssociation::send_pdata(self, presentation_context_id)
    }

    /// Prepare a P-Data reader for receiving
    /// one or more data item PDUs.
    ///
    /// Returns a reader which automatically
    /// receives more data PDUs once the bytes collected are consumed.
    pub fn receive_pdata(&mut self) -> crate::association::pdata::PDataReader<'_, &mut S> {
        SyncAssociation::receive_pdata(self)
    }

    /// Obtain access to the inner stream
    /// connected to the association acceptor.
    ///
    /// This can be used to send the PDU in semantic fragments of the message,
    /// thus using less memory.
    ///
    /// **Note:** reading and writing should be done with care
    /// to avoid inconsistencies in the association state.
    /// Do not call `send` and `receive` while not in a PDU boundary.
    pub fn inner_stream(&mut self) -> &mut S {
        SyncAssociation::inner_stream(self)
    }
}

impl<S> Association for ServerAssociation<S>
where
    S: std::io::Read + std::io::Write + CloseSocket,
{
    /// Obtain a view of the negotiated presentation contexts.
    fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
        &self.presentation_contexts
    }

    /// 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 the remote DICOM node's application entity title.
    fn peer_ae_title(&self) -> &str {
        &self.client_ae_title
    }

    /// Retrieve the user variables that were taken from the server.
    ///
    /// It usually contains the maximum PDU length,
    /// the implementation class UID, and the implementation version name.
    fn user_variables(&self) -> &[UserVariableItem] {
        &self.user_variables
    }
}

impl<S> SyncAssociationSealed<S> for ServerAssociation<S>
where
    S: std::io::Read + std::io::Write + CloseSocket,
{
    fn send(&mut self, pdu: &Pdu) -> Result<()> {
        self.write_buffer.clear();
        encode_pdu(
            &mut self.write_buffer,
            pdu,
            self.requestor_max_pdu_length + PDU_HEADER_SIZE,
        )?;
        self.socket
            .write_all(&self.write_buffer)
            .context(WireSendSnafu)
    }

    fn receive(&mut self) -> Result<Pdu> {
        read_pdu_from_wire(
            &mut self.socket,
            &mut self.read_buffer,
            self.acceptor_max_pdu_length,
            self.strict,
        )
    }

    fn close(&mut self) -> std::io::Result<()> {
        self.socket.close()
    }
}

impl<S> SyncAssociation<S> for ServerAssociation<S>
where
    S: std::io::Read + std::io::Write + CloseSocket,
{
    fn inner_stream(&mut self) -> &mut S {
        &mut self.socket
    }

    fn get_mut(&mut self) -> (&mut S, &mut BytesMut) {
        let Self {
            socket,
            read_buffer,
            ..
        } = self;
        (socket, read_buffer)
    }
}

/// Check that a transfer syntax repository
/// supports the given transfer syntax,
/// meaning that it can parse and decode DICOM data sets.
///
/// ```
/// # use dicom_transfer_syntax_registry::TransferSyntaxRegistry;
/// # use dicom_ul::association::server::is_supported_with_repo;
/// // Implicit VR Little Endian is guaranteed to be supported
/// assert!(is_supported_with_repo(TransferSyntaxRegistry, "1.2.840.10008.1.2"));
/// ```
pub fn is_supported_with_repo<R>(ts_repo: R, ts_uid: &str) -> bool
where
    R: TransferSyntaxIndex,
{
    ts_repo
        .get(ts_uid)
        .filter(|ts| !ts.is_unsupported())
        .is_some()
}

/// Check that the main transfer syntax registry
/// supports the given transfer syntax,
/// meaning that it can parse and decode DICOM data sets.
///
/// ```
/// # use dicom_ul::association::server::is_supported;
/// // Implicit VR Little Endian is guaranteed to be supported
/// assert!(is_supported("1.2.840.10008.1.2"));
/// ```
pub fn is_supported(ts_uid: &str) -> bool {
    is_supported_with_repo(TransferSyntaxRegistry, ts_uid)
}

/// From a sequence of transfer syntaxes,
/// choose the first transfer syntax to be supported
/// by the given transfer syntax repository.
pub fn choose_supported_with_repo<R, I, T>(ts_repo: R, it: I) -> Option<T>
where
    R: TransferSyntaxIndex,
    I: IntoIterator<Item = T>,
    T: AsRef<str>,
{
    it.into_iter()
        .find(|ts| is_supported_with_repo(&ts_repo, ts.as_ref()))
}

/// From a sequence of transfer syntaxes,
/// choose the first transfer syntax to be supported
/// by the main transfer syntax registry.
pub fn choose_supported<I, T>(it: I) -> Option<T>
where
    I: IntoIterator<Item = T>,
    T: AsRef<str>,
{
    it.into_iter().find(|ts| is_supported(ts.as_ref()))
}

#[cfg(feature = "async-tls")]
async fn accept_tls_async<S>(socket: S) -> Result<tokio_rustls::server::StartHandshake<S>>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin,
{
    tokio_rustls::LazyConfigAcceptor::new(rustls::server::Acceptor::default(), socket)
        .await
        .context(super::TlsHandshakeSnafu)
}

#[cfg(feature = "async-tls")]
async fn handshake_tls_async(
    socket: tokio::net::TcpStream,
    tls_config: Arc<ServerConfig>,
) -> Result<tokio_rustls::server::TlsStream<tokio::net::TcpStream>> {
    let accepted = accept_tls_async(socket).await?;
    let stream = accepted
        .into_stream(tls_config)
        .await
        .context(super::TlsHandshakeSnafu)?;
    Ok(stream)
}

#[cfg(feature = "async")]
impl<A, N> ServerAssociationOptions<'_, A, N>
where
    A: AccessControl,
    N: Negotiation,
{
    /// Negotiate an association with the given TCP stream.
    pub async fn establish_async(
        &self,
        mut socket: tokio::net::TcpStream,
    ) -> Result<AsyncServerAssociation<tokio::net::TcpStream>> {
        use tokio::io::AsyncWriteExt;
        ensure!(
            !self.abstract_syntax_uids.is_empty() || self.promiscuous,
            MissingAbstractSyntaxSnafu
        );
        let read_timeout = self.socket_options.read_timeout;
        let task = async {
            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let pdu = match super::read_pdu_from_wire_async(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )
            .await
            {
                Ok(pdu) => pdu,
                Err(e) => {
                    // Attempt to check if the failure was because the client sent a TLS stream
                    #[cfg(feature = "async-tls")]
                    {
                        // `read_pdu_from_wire_async` pulls bytes off the socket
                        // and accumulates into `read_buffer`, so we need to use that in
                        // `accept_tls_async`

                        // NOTE: `accept_tls_async` may attempt to write to the cursor if
                        // an `AcceptedAlert` is generated, this should be fine since the
                        // client does not need to receive that alert message
                        let cursor = std::io::Cursor::new(read_buffer.to_vec());
                        if accept_tls_async(cursor).await.is_ok() {
                            return super::TlsNotSupportedSnafu.fail();
                        }
                    }
                    return Err(e);
                }
            };

            let mut write_buffer: Vec<u8> =
                Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
            match self.process_a_association_rq(pdu) {
                Ok((
                    pdu,
                    NegotiatedOptions {
                        user_variables,
                        presentation_contexts,
                        peer_max_pdu_length,
                        peer_ae_title,
                    },
                    called_ae_title,
                )) => {
                    write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                    socket
                        .write_all(&write_buffer)
                        .await
                        .context(WireSendSnafu)?;
                    Ok(AsyncServerAssociation {
                        presentation_contexts,
                        requestor_max_pdu_length: peer_max_pdu_length,
                        acceptor_max_pdu_length: self.max_pdu_length,
                        socket,
                        client_ae_title: peer_ae_title,
                        write_buffer,
                        strict: self.strict,
                        read_buffer,
                        read_timeout: self.socket_options.read_timeout,
                        write_timeout: self.socket_options.write_timeout,
                        user_variables,
                        called_ae_title,
                    })
                }
                Err((pdu, err)) => {
                    // send the rejection/abort PDU
                    write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                    socket
                        .write_all(&write_buffer)
                        .await
                        .context(WireSendSnafu)?;
                    Err(err)
                }
            }
        };
        super::timeout(read_timeout, task).await
    }

    /// Negotiate an association with the given TCP stream.
    #[cfg(feature = "async-tls")]
    pub async fn establish_tls_async(
        &self,
        socket: tokio::net::TcpStream,
    ) -> Result<AsyncServerAssociation<AsyncTlsStream>> {
        use tokio::io::AsyncWriteExt;

        ensure!(
            !self.abstract_syntax_uids.is_empty() || self.promiscuous,
            MissingAbstractSyntaxSnafu
        );
        let tls_config = self
            .tls_config
            .as_ref()
            .ok_or_else(|| crate::association::TlsConfigMissingSnafu {}.build())?;
        let mut socket = handshake_tls_async(socket, tls_config.clone()).await?;
        let read_timeout = self.socket_options.read_timeout;
        let task = async {
            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let pdu = super::read_pdu_from_wire_async(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )
            .await?;

            let mut write_buffer: Vec<u8> =
                Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
            match self.process_a_association_rq(pdu) {
                Ok((
                    pdu,
                    NegotiatedOptions {
                        user_variables,
                        presentation_contexts,
                        peer_max_pdu_length,
                        peer_ae_title,
                    },
                    called_ae_title,
                )) => {
                    write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                    socket
                        .write_all(&write_buffer)
                        .await
                        .context(WireSendSnafu)?;
                    Ok(AsyncServerAssociation {
                        presentation_contexts,
                        requestor_max_pdu_length: peer_max_pdu_length,
                        acceptor_max_pdu_length: self.max_pdu_length,
                        socket,
                        client_ae_title: peer_ae_title,
                        write_buffer,
                        strict: self.strict,
                        read_buffer,
                        read_timeout: self.socket_options.read_timeout,
                        write_timeout: self.socket_options.write_timeout,
                        user_variables,
                        called_ae_title,
                    })
                }
                Err((pdu, err)) => {
                    // send the rejection/abort PDU
                    write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
                    socket
                        .write_all(&write_buffer)
                        .await
                        .context(WireSendSnafu)?;
                    Err(err)
                }
            }
        };
        super::timeout(read_timeout, task).await
    }
}

/// An async DICOM upper level association from the perspective
/// of an accepting application entity.
///
/// The most common operations of an established association are
/// [`send`](crate::association::AsyncAssociation::send)
/// and [`receive`](crate::association::AsyncAssociation::receive).
/// Sending large P-Data fragments may be easier through the P-Data sender
/// abstraction (see [`send_pdata`](crate::association::AsyncAssociation::send_pdata)).
///
/// When the value falls out of scope,
/// the program will shut down the underlying TCP connection.
#[cfg(feature = "async")]
#[derive(Debug)]
pub struct AsyncServerAssociation<S> {
    /// The accorded presentation contexts
    presentation_contexts: Vec<PresentationContextNegotiated>,
    /// The maximum PDU length that the remote application entity accepts
    requestor_max_pdu_length: u32,
    /// The maximum PDU length that this application entity is expecting to receive
    acceptor_max_pdu_length: u32,
    /// The TCP stream to the other DICOM node
    socket: S,
    /// The application entity title of the other DICOM node
    client_ae_title: String,
    /// The AE title the calling client used.
    called_ae_title: String,
    /// write buffer to send fully assembled PDUs on wire
    write_buffer: Vec<u8>,
    /// whether to receive PDUs in strict mode
    strict: bool,
    /// Read buffer from the socket
    read_buffer: bytes::BytesMut,
    /// Timeout for individual receive operations
    read_timeout: Option<std::time::Duration>,
    /// Timeout for individual send operations
    write_timeout: Option<std::time::Duration>,
    /// User variables received from the peer
    user_variables: Vec<UserVariableItem>,
}

#[cfg(feature = "async")]
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
    }

    /// Obtain the remote DICOM node's application entity title.
    fn peer_ae_title(&self) -> &str {
        &self.client_ae_title
    }

    fn user_variables(&self) -> &[UserVariableItem] {
        &self.user_variables
    }
}

#[cfg(feature = "async")]
impl<S> crate::association::private::AsyncAssociationSealed<S> for AsyncServerAssociation<S>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
{
    /// Send a PDU message to the other intervenient.
    async fn send(&mut self, msg: &Pdu) -> Result<()> {
        use tokio::io::AsyncWriteExt;
        self.write_buffer.clear();
        super::timeout(self.write_timeout, async {
            encode_pdu(
                &mut self.write_buffer,
                msg,
                self.requestor_max_pdu_length + PDU_HEADER_SIZE,
            )?;
            self.socket
                .write_all(&self.write_buffer)
                .await
                .context(WireSendSnafu)
        })
        .await
    }

    /// Read a PDU message from the other intervenient.
    async fn receive(&mut self) -> Result<Pdu> {
        super::timeout(self.read_timeout, async {
            super::read_pdu_from_wire_async(
                &mut self.socket,
                &mut self.read_buffer,
                self.acceptor_max_pdu_length,
                self.strict,
            )
            .await
        })
        .await
    }

    async fn close(&mut self) -> std::io::Result<()> {
        use tokio::io::AsyncWriteExt;
        self.socket.shutdown().await
    }
}

#[cfg(feature = "async")]
impl<S> crate::association::AsyncAssociation<S> for AsyncServerAssociation<S>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
{
    fn inner_stream(&mut self) -> &mut S {
        &mut self.socket
    }

    fn get_mut(&mut self) -> (&mut S, &mut bytes::BytesMut) {
        let Self {
            socket,
            read_buffer,
            ..
        } = self;
        (socket, read_buffer)
    }
}

// compatibility filler, remove in 0.10.0
#[cfg(feature = "async")]
impl<S> AsyncServerAssociation<S> {
    /// Obtain a view of the negotiated presentation contexts.
    pub fn presentation_contexts(&self) -> &[PresentationContextNegotiated] {
        &self.presentation_contexts
    }

    /// Retrieve the maximum PDU length
    /// that the association acceptor is expecting to receive.
    pub 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.
    pub fn requestor_max_pdu_length(&self) -> u32 {
        self.requestor_max_pdu_length
    }

    /// Obtain the remote DICOM node's application entity title.
    #[deprecated(
        since = "0.9.1",
        note = "Call `peer_ae_title` from trait `Association`"
    )]
    pub fn client_ae_title(&self) -> &str {
        &self.client_ae_title
    }

    /// Obtain the application entity title called by the client.
    pub fn called_ae_title(&self) -> &str {
        &self.called_ae_title
    }
}

// compatibility filler, remove in 0.10.0
#[cfg(feature = "async")]
impl<S> AsyncServerAssociation<S>
where
    S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send,
{
    /// Send a PDU message to the other intervenient.
    pub async fn send(&mut self, msg: &Pdu) -> Result<()> {
        AsyncAssociation::send(self, msg).await
    }

    /// Read a PDU message from the other intervenient.
    pub async fn receive(&mut self) -> Result<Pdu> {
        AsyncAssociation::receive(self).await
    }

    /// Iniate a graceful release of the association.
    ///
    /// A DIMSE A-RELEASE transaction is initiated by this application entity,
    /// and the underlying socket is closed once settled.
    ///
    /// Note that implementers of this trait
    /// do not try to release the association on [`Drop`],
    /// so remember to call `release` explicitly
    /// at the end of all DIMSE transactions.
    pub async fn release(self) -> Result<()> {
        AsyncAssociation::release(self).await
    }

    /// Send a provider initiated abort message
    /// and shut down the TCP connection,
    /// terminating the association.
    pub async fn abort(self) -> Result<()> {
        AsyncAssociation::abort(self).await
    }

    /// Prepare a P-Data writer for sending
    /// one or more data item PDUs.
    ///
    /// Returns a writer which automatically
    /// splits the inner data into separate PDUs if necessary.
    pub fn send_pdata(
        &mut self,
        presentation_context_id: u8,
    ) -> crate::association::pdata::non_blocking::AsyncPDataWriter<&mut S> {
        AsyncAssociation::send_pdata(self, presentation_context_id)
    }

    /// Prepare a P-Data reader for receiving
    /// one or more data item PDUs.
    ///
    /// Returns a reader which automatically
    /// receives more data PDUs once the bytes collected are consumed.
    pub fn receive_pdata(&mut self) -> crate::association::pdata::PDataReader<'_, &mut S> {
        AsyncAssociation::receive_pdata(self)
    }

    /// Obtain access to the inner stream
    /// connected to the association acceptor.
    ///
    /// This can be used to send the PDU in semantic fragments of the message,
    /// thus using less memory.
    ///
    /// **Note:** reading and writing should be done with care
    /// to avoid inconsistencies in the association state.
    /// Do not call `send` and `receive` while not in a PDU boundary.
    pub fn inner_stream(&mut self) -> &mut S {
        AsyncAssociation::inner_stream(self)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_choose_supported() {
        assert_eq!(choose_supported(vec!["1.1.1.1.1"]), None,);

        // string slices, impl VR first
        assert_eq!(
            choose_supported(vec!["1.2.840.10008.1.2", "1.2.840.10008.1.2.1"]),
            Some("1.2.840.10008.1.2"),
        );

        // heap allocated strings slices, expl VR first
        assert_eq!(
            choose_supported(vec![
                "1.2.840.10008.1.2.1".to_string(),
                "1.2.840.10008.1.2".to_string()
            ]),
            Some("1.2.840.10008.1.2.1".to_string()),
        );
    }

    impl<'a, A, N> ServerAssociationOptions<'a, A, N>
    where
        A: AccessControl,
        N: Negotiation,
    {
        // Broken implementation of server establish which sends an extra pdu during establish
        pub(crate) fn establish_with_extra_pdus(
            &self,
            mut socket: std::net::TcpStream,
            extra_pdus: Vec<Pdu>,
        ) -> Result<ServerAssociation<TcpStream>> {
            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let pdu = read_pdu_from_wire(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )?;
            let (
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            ) = self
                .process_a_association_rq(pdu)
                .expect("Could not parse association req");

            let mut write_buffer: Vec<u8> =
                Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
            write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
            for extra_pdu in extra_pdus {
                write_pdu(&mut write_buffer, &extra_pdu).context(SendPduSnafu)?;
            }
            socket.write_all(&write_buffer).context(WireSendSnafu)?;

            Ok(ServerAssociation {
                presentation_contexts,
                requestor_max_pdu_length: peer_max_pdu_length,
                acceptor_max_pdu_length: self.max_pdu_length,
                socket,
                client_ae_title: peer_ae_title,
                write_buffer,
                read_buffer,
                strict: self.strict,
                user_variables,
                called_ae_title,
            })
        }

        // Broken implementation of server establish which sends an extra pdu during establish
        #[cfg(feature = "async")]
        pub(crate) async fn establish_with_extra_pdus_async(
            &self,
            mut socket: tokio::net::TcpStream,
            extra_pdus: Vec<Pdu>,
        ) -> Result<AsyncServerAssociation<tokio::net::TcpStream>> {
            use tokio::io::AsyncWriteExt;

            use crate::association::read_pdu_from_wire_async;

            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let pdu = read_pdu_from_wire_async(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )
            .await?;
            let (
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            ) = self
                .process_a_association_rq(pdu)
                .expect("Could not parse association req");

            let mut buffer: Vec<u8> = Vec::with_capacity(
                (peer_max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            write_pdu(&mut buffer, &pdu).context(SendPduSnafu)?;
            for extra_pdu in extra_pdus {
                write_pdu(&mut buffer, &extra_pdu).context(SendPduSnafu)?;
            }
            socket.write_all(&buffer).await.context(WireSendSnafu)?;

            Ok(AsyncServerAssociation {
                presentation_contexts,
                requestor_max_pdu_length: peer_max_pdu_length,
                acceptor_max_pdu_length: self.max_pdu_length,
                socket,
                client_ae_title: peer_ae_title,
                write_buffer: buffer,
                strict: self.strict,
                read_buffer: BytesMut::with_capacity(
                    (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
                ),
                user_variables,
                read_timeout: self.socket_options.read_timeout,
                write_timeout: self.socket_options.write_timeout,
                called_ae_title,
            })
        }

        // Broken implementation of server establish which reproduces behavior that #589 introduced
        pub fn broken_establish(
            &self,
            mut socket: TcpStream,
        ) -> Result<ServerAssociation<TcpStream>> {
            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let msg = read_pdu_from_wire(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )?;
            let (
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            ) = self
                .process_a_association_rq(msg)
                .expect("Could not parse association req");
            let mut write_buffer: Vec<u8> =
                Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
            write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
            socket.write_all(&write_buffer).context(WireSendSnafu)?;
            read_buffer.clear();
            Ok(ServerAssociation {
                presentation_contexts,
                requestor_max_pdu_length: peer_max_pdu_length,
                acceptor_max_pdu_length: self.max_pdu_length,
                socket,
                client_ae_title: peer_ae_title,
                write_buffer,
                strict: self.strict,
                read_buffer,
                user_variables,
                called_ae_title,
            })
        }

        // Broken implementation of server establish which reproduces behavior that #589 introduced
        #[cfg(feature = "async")]
        pub async fn broken_establish_async(
            &self,
            mut socket: tokio::net::TcpStream,
        ) -> Result<AsyncServerAssociation<tokio::net::TcpStream>> {
            use tokio::io::AsyncWriteExt;

            use crate::association::read_pdu_from_wire_async;

            let mut read_buffer = BytesMut::with_capacity(
                (self.max_pdu_length.min(LARGE_PDU_SIZE) + PDU_HEADER_SIZE) as usize,
            );
            let msg = read_pdu_from_wire_async(
                &mut socket,
                &mut read_buffer,
                self.max_pdu_length,
                self.strict,
            )
            .await?;
            let (
                pdu,
                NegotiatedOptions {
                    user_variables,
                    presentation_contexts,
                    peer_max_pdu_length,
                    peer_ae_title,
                },
                called_ae_title,
            ) = self
                .process_a_association_rq(msg)
                .expect("Could not parse association req");
            let mut write_buffer: Vec<u8> =
                Vec::with_capacity((DEFAULT_MAX_PDU + PDU_HEADER_SIZE) as usize);
            write_pdu(&mut write_buffer, &pdu).context(SendPduSnafu)?;
            socket
                .write_all(&write_buffer)
                .await
                .context(WireSendSnafu)?;
            read_buffer.clear();
            Ok(AsyncServerAssociation {
                presentation_contexts,
                requestor_max_pdu_length: peer_max_pdu_length,
                acceptor_max_pdu_length: self.max_pdu_length,
                socket,
                client_ae_title: peer_ae_title,
                write_buffer,
                strict: self.strict,
                read_buffer,
                read_timeout: self.socket_options.read_timeout,
                write_timeout: self.socket_options.write_timeout,
                user_variables,
                called_ae_title,
            })
        }
    }
}