ramqp 0.6.5

A from-scratch, clean-room AMQP 1.0 client.
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

//! Sender link state (WP-4.1). The send orchestration (credit/window gating,
//! multi-frame splitting, disposition resolution) lives in the session, which
//! owns the transport; this struct holds the per-link state it operates on.

use std::collections::{HashMap, VecDeque};

use bytes::Bytes;
use tokio::sync::mpsc;

use crate::config::CreditMode;
use crate::error::{LinkError, SendError};
use crate::link::credit::LinkCredit;
use crate::link::settlement::UnsettledMap;
use crate::proto::{LinkAttached, LinkEvent, Reply};
use crate::types::definitions::SenderSettleMode;
use crate::types::messaging::DeliveryState;

/// Safety cap on outstanding unsettled deliveries per sender link. Unsettled
/// sends pause (messages stay queued) once this many deliveries await a
/// disposition, bounding memory even if a broker grants credit but withholds
/// settlement. Well above any realistic in-flight count for normal operation.
pub const MAX_UNSETTLED_PER_LINK: usize = 1 << 16;

/// A queued outbound message awaiting credit/window before it can be sent.
#[derive(Debug)]
pub struct PendingSend {
    /// The pre-encoded message bytes.
    pub body: Bytes,
    /// Whether to send pre-settled.
    pub settled: bool,
    /// The message format.
    pub message_format: u32,
    /// Delivery state to set on the transfer (e.g. a `transactional-state`).
    pub state: Option<DeliveryState>,
    /// Reply with the terminal outcome once settled (if awaited).
    pub reply: Option<Reply<DeliveryState, SendError>>,
}

/// Per-sender-link state owned by the session.
#[derive(Debug)]
pub struct SenderLink {
    /// Our local handle.
    pub handle: u32,
    /// The peer's handle (from its responding attach).
    pub remote_handle: Option<u32>,
    /// The link name.
    pub name: String,
    /// Whether the link is attached.
    pub attached: bool,
    /// Channel to the producer handle.
    pub events: mpsc::Sender<LinkEvent>,
    /// The attach reply, held until the peer responds.
    pub pending_attach: Option<Reply<LinkAttached, LinkError>>,
    /// Credit / flow state.
    pub credit: LinkCredit,
    /// Unsettled (sent, awaiting disposition) deliveries.
    pub unsettled: UnsettledMap,
    /// Per-delivery settle-awaiting replies, with the send instant (for the
    /// send-to-settle latency metric).
    pub pending: HashMap<u32, (Reply<DeliveryState, SendError>, std::time::Instant)>,
    /// Messages queued because there was no credit/window when sent.
    pub outbox: VecDeque<PendingSend>,
    /// Our requested sender settle mode.
    pub settle_mode: SenderSettleMode,
    next_tag: u64,
}

impl SenderLink {
    /// Create a sender link in the unattached state.
    pub fn new(
        handle: u32,
        name: String,
        events: mpsc::Sender<LinkEvent>,
        pending_attach: Reply<LinkAttached, LinkError>,
        settle_mode: SenderSettleMode,
        credit_mode: CreditMode,
    ) -> Self {
        SenderLink {
            handle,
            remote_handle: None,
            name,
            attached: false,
            events,
            pending_attach: Some(pending_attach),
            credit: LinkCredit::new(0, credit_mode),
            unsettled: UnsettledMap::new(),
            pending: HashMap::new(),
            outbox: VecDeque::new(),
            settle_mode,
            next_tag: 0,
        }
    }

    /// Allocate the next delivery tag (an 8-byte big-endian counter).
    pub fn next_delivery_tag(&mut self) -> Bytes {
        let tag = self.next_tag;
        self.next_tag = self.next_tag.wrapping_add(1);
        Bytes::copy_from_slice(&tag.to_be_bytes())
    }
}