wireframe 0.3.0

Simplify building servers and clients for custom binary protocols.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

//! Public adapter contract for transport-level fragmentation and reassembly.
//!
//! [`FragmentAdapter`] captures the minimal behaviour required to split outbound
//! packets into transport fragments, rebuild inbound fragments into full
//! packets, and purge stale partial assemblies.

use bincode::error::DecodeError;
use thiserror::Error;

use super::{
    Fragmentable,
    FragmentationConfig,
    FragmentationError,
    Fragmenter,
    MessageId,
    Reassembler,
    ReassemblyError,
    decode_fragment_payload,
    fragment_packet,
    packet::FragmentParts,
};

/// Error returned by [`FragmentAdapter::reassemble`].
#[derive(Debug, Error)]
pub enum FragmentAdapterError {
    /// Fragment payload marker/header decoding failed.
    #[error("decode error: {0}")]
    Decode(#[from] DecodeError),
    /// Reassembly ordering or size validation failed.
    #[error("reassembly error: {0}")]
    Reassembly(#[from] ReassemblyError),
}

/// Adapter contract for transport-level fragmentation and reassembly.
pub trait FragmentAdapter: Send + Sync {
    /// Attempt to fragment a packet for outbound transport.
    ///
    /// # Errors
    ///
    /// Returns [`FragmentationError`] when payload chunking or header encoding
    /// fails.
    fn fragment<E: Fragmentable>(&self, packet: E) -> Result<Vec<E>, FragmentationError>;

    /// Attempt to reassemble an inbound packet.
    ///
    /// Returns `Ok(Some(packet))` when a complete packet is available,
    /// `Ok(None)` when more fragments are required, and an error when decoding
    /// or reassembly invariants fail.
    ///
    /// # Errors
    ///
    /// Returns [`FragmentAdapterError`] when fragment decoding fails or when
    /// ordering and size guarantees are violated.
    fn reassemble<E: Fragmentable>(&mut self, packet: E)
    -> Result<Option<E>, FragmentAdapterError>;

    /// Purge stale partial reassembly state.
    ///
    /// Returns the identifiers that were evicted.
    fn purge_expired(&mut self) -> Vec<MessageId>;
}

/// Default adapter backed by [`Fragmenter`] and [`Reassembler`].
#[derive(Debug)]
pub struct DefaultFragmentAdapter {
    fragmenter: Fragmenter,
    reassembler: Reassembler,
}

impl DefaultFragmentAdapter {
    /// Create a default adapter from fragmentation configuration.
    #[must_use]
    pub fn new(config: FragmentationConfig) -> Self {
        Self {
            fragmenter: Fragmenter::new(config.fragment_payload_cap),
            reassembler: Reassembler::new(config.max_message_size, config.reassembly_timeout),
        }
    }

    /// Fragment outbound packet data.
    ///
    /// # Errors
    ///
    /// Returns [`FragmentationError`] when fragment emission fails.
    pub fn fragment<E: Fragmentable>(&self, packet: E) -> Result<Vec<E>, FragmentationError> {
        fragment_packet(&self.fragmenter, packet)
    }

    /// Reassemble inbound packet data.
    ///
    /// # Errors
    ///
    /// Returns [`FragmentAdapterError`] when decoding or reassembly fails.
    pub fn reassemble<E: Fragmentable>(
        &mut self,
        packet: E,
    ) -> Result<Option<E>, FragmentAdapterError> {
        let parts = packet.into_fragment_parts();
        let id = parts.id();
        let correlation_id = parts.correlation_id();
        let payload = parts.into_payload();

        if let Some((header, fragment_payload)) = decode_fragment_payload(&payload)? {
            match self.reassembler.push(header, fragment_payload)? {
                Some(message) => {
                    let rebuilt = FragmentParts::new(id, correlation_id, message.into_payload());
                    Ok(Some(E::from_fragment_parts(rebuilt)))
                }
                None => Ok(None),
            }
        } else {
            let passthrough = FragmentParts::new(id, correlation_id, payload);
            Ok(Some(E::from_fragment_parts(passthrough)))
        }
    }

    /// Purge stale reassembly entries and return evicted identifiers.
    pub fn purge_expired(&mut self) -> Vec<MessageId> { self.reassembler.purge_expired() }
}

impl FragmentAdapter for DefaultFragmentAdapter {
    fn fragment<E: Fragmentable>(&self, packet: E) -> Result<Vec<E>, FragmentationError> {
        DefaultFragmentAdapter::fragment(self, packet)
    }

    fn reassemble<E: Fragmentable>(
        &mut self,
        packet: E,
    ) -> Result<Option<E>, FragmentAdapterError> {
        DefaultFragmentAdapter::reassemble(self, packet)
    }

    fn purge_expired(&mut self) -> Vec<MessageId> { DefaultFragmentAdapter::purge_expired(self) }
}