base_endpoint/
composite_endpoint.rs

1// Released under the MIT License.
2// Copyright, 2026, by Samuel Williams.
3
4use crate::{BindEndpoint, BindOptions, ConnectEndpoint, no_addresses, rollback};
5use base_scheduler::{Cancellation, OperationError, OperationResult, SocketClose};
6
7/// An ordered collection of endpoints of one type. Connecting tries children in order until one succeeds. Binding creates every child's listeners and rolls them all back if a later child fails. Empty collections fail with `AddrNotAvailable`.
8#[derive(Clone, Debug, Eq, PartialEq)]
9pub struct CompositeEndpoint<E> {
10    endpoints: Vec<E>,
11}
12
13impl<E> CompositeEndpoint<E> {
14    /// Describe an ordered collection, without performing I/O.
15    pub fn new(endpoints: Vec<E>) -> Self {
16        Self { endpoints }
17    }
18
19    /// The child endpoint descriptions in attempt order.
20    pub fn endpoints(&self) -> &[E] {
21        &self.endpoints
22    }
23}
24
25impl<S, E> ConnectEndpoint<S> for CompositeEndpoint<E>
26where
27    S: Sync + ?Sized,
28    E: ConnectEndpoint<S>,
29{
30    type Connection = E::Connection;
31
32    async fn connect(
33        &self,
34        scheduler: &S,
35        cancellation: &Cancellation,
36    ) -> OperationResult<Self::Connection> {
37        cancellation.check()?;
38        let mut last_error = None;
39        for endpoint in &self.endpoints {
40            cancellation.check()?;
41            match endpoint.connect(scheduler, cancellation).await {
42                Ok(connection) => return Ok(connection),
43                Err(error @ OperationError::Cancelled(_)) => return Err(error),
44                Err(error) => last_error = Some(error),
45            }
46        }
47        Err(last_error.unwrap_or_else(no_addresses))
48    }
49}
50
51// Binding composition currently supports listeners closed by the scheduler. Connect composition has no socket representation requirement.
52impl<S, E> BindEndpoint<S> for CompositeEndpoint<E>
53where
54    S: SocketClose<Socket = E::Listener> + ?Sized,
55    E: BindEndpoint<S>,
56{
57    type Listener = E::Listener;
58
59    async fn bind(
60        &self,
61        scheduler: &S,
62        options: BindOptions,
63        cancellation: &Cancellation,
64    ) -> OperationResult<Vec<Self::Listener>> {
65        cancellation.check()?;
66        if self.endpoints.is_empty() {
67            return Err(no_addresses());
68        }
69        let mut listeners = Vec::new();
70        for endpoint in &self.endpoints {
71            let result = match cancellation.check() {
72                Ok(()) => endpoint.bind(scheduler, options, cancellation).await,
73                Err(error) => Err(error.into()),
74            };
75            match result {
76                Ok(mut bound) => listeners.append(&mut bound),
77                Err(error) => {
78                    rollback(scheduler, listeners).await;
79                    return Err(error);
80                }
81            }
82        }
83        Ok(listeners)
84    }
85}
base_endpoint/composite_endpoint.rs

base_endpoint/
composite_endpoint.rs
