brokapi_runtime/connection.rs
1//! Connection desired/actual subscription tracking.
2
3use std::collections::HashMap;
4use std::time::Instant;
5
6use brokapi_sansio::{DesiredSubscription, SubscriptionSet};
7
8use crate::health::{ConnectionHealth, HealthClass};
9
10/// What the connection knows about one subscription: the "actual" side of desired-versus-actual.
11#[derive(Debug, Clone, PartialEq, Eq)]
12pub enum SubState {
13 /// The subscribe request was sent and no answer has arrived yet.
14 Pending {
15 /// When the request was sent. The runtime does not time out pending subscriptions by
16 /// itself.
17 since: Instant,
18 },
19 /// The broker confirmed the subscription.
20 Active {
21 /// Sequence number of the last message seen for it, as supplied by the owner.
22 seq: u64,
23 },
24 /// The broker refused the subscription or it broke.
25 Failed {
26 /// Human-readable cause for logs.
27 reason: String,
28 },
29}
30
31/// The desired and actual subscriptions of one connection, and its health.
32///
33/// The owner sets what it wants ([`set_desired`](Self::set_desired)) and records what happens on
34/// the wire ([`mark_pending`](Self::mark_pending), [`mark_active`](Self::mark_active),
35/// [`mark_failed`](Self::mark_failed)). The difference between the two is what
36/// [`needs_resubscribe`](Self::needs_resubscribe) and [`needs_unsubscribe`](Self::needs_unsubscribe)
37/// report. The struct does no I/O.
38///
39/// # Examples
40///
41/// ```
42/// use std::time::Instant;
43/// use brokapi_core::{InstrumentId, InstrumentKey, NativeId, SubscriptionId};
44/// use brokapi_runtime::ConnectionState;
45/// use brokapi_sansio::{DesiredSubscription, SubscriptionKind, SubscriptionSet};
46///
47/// let sber = InstrumentId::new(
48/// InstrumentKey::new("SBER", "TQBR"),
49/// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
50/// );
51/// let sub = DesiredSubscription {
52/// id: SubscriptionId::new(1),
53/// kind: SubscriptionKind::Quotes { instrument: sber },
54/// };
55///
56/// let mut conn = ConnectionState::new(1, Instant::now());
57/// conn.set_desired(SubscriptionSet::with_subscriptions(vec![sub.clone()]));
58/// // Wanted but not yet on the wire.
59/// assert_eq!(conn.needs_resubscribe().len(), 1);
60///
61/// conn.mark_active(sub.id, 1);
62/// assert!(conn.needs_resubscribe().is_empty());
63///
64/// // The broker drops it: it must be sent again.
65/// conn.mark_failed(sub.id, "denied");
66/// assert_eq!(conn.needs_resubscribe().len(), 1);
67/// ```
68#[derive(Debug)]
69pub struct ConnectionState {
70 /// Connection id, unique inside a [`Supervisor`](crate::Supervisor).
71 pub id: u64,
72 /// The subscriptions the owner wants on this connection.
73 pub desired: SubscriptionSet,
74 /// What is known about each subscription on the wire.
75 pub actual: HashMap<brokapi_core::SubscriptionId, SubState>,
76 /// Receive-side health of the connection.
77 pub health: ConnectionHealth,
78}
79
80impl ConnectionState {
81 /// Creates a connection with nothing desired and nothing active, last heard from at `now`.
82 pub fn new(id: u64, now: Instant) -> Self {
83 Self {
84 id,
85 desired: SubscriptionSet::new(),
86 actual: HashMap::new(),
87 health: ConnectionHealth::new(now),
88 }
89 }
90
91 /// Replaces the whole desired set. The `actual` map is left as it is.
92 pub fn set_desired(&mut self, desired: SubscriptionSet) {
93 self.desired = desired;
94 }
95
96 /// Records that the subscribe request for `sub` was sent at `now`.
97 pub fn mark_pending(&mut self, sub: &DesiredSubscription, now: Instant) {
98 self.actual.insert(sub.id, SubState::Pending { since: now });
99 }
100
101 /// Records that the broker confirmed `sub_id`, with the sequence number `seq`.
102 pub fn mark_active(&mut self, sub_id: brokapi_core::SubscriptionId, seq: u64) {
103 self.actual.insert(sub_id, SubState::Active { seq });
104 }
105
106 /// Records that `sub_id` was refused or broke, with a cause for logs.
107 pub fn mark_failed(&mut self, sub_id: brokapi_core::SubscriptionId, reason: impl Into<String>) {
108 self.actual.insert(
109 sub_id,
110 SubState::Failed {
111 reason: reason.into(),
112 },
113 );
114 }
115
116 /// Forgets the actual state of `sub_id`. It does not touch the desired set.
117 pub fn remove(&mut self, sub_id: brokapi_core::SubscriptionId) {
118 self.actual.remove(&sub_id);
119 }
120
121 /// Classifies the connection health, see [`ConnectionHealth::classify`].
122 pub fn classify(
123 &self,
124 now: Instant,
125 dead_after: std::time::Duration,
126 degraded_after: std::time::Duration,
127 ) -> HealthClass {
128 self.health.classify(now, dead_after, degraded_after)
129 }
130
131 /// Desired subscriptions that are not confirmed active, to be (re)sent without a full reconnect.
132 ///
133 /// A subscription is listed when it has no state, has failed, or is still pending. Pending ones
134 /// are included whatever their age, so call this when you are ready to send again, not on every
135 /// tick, or you will resend requests that are merely slow.
136 pub fn needs_resubscribe(&self) -> Vec<DesiredSubscription> {
137 self.desired
138 .subs()
139 .iter()
140 .filter(|sub| match self.actual.get(&sub.id) {
141 None | Some(SubState::Failed { .. }) | Some(SubState::Pending { .. }) => true,
142 Some(SubState::Active { .. }) => false,
143 })
144 .cloned()
145 .collect()
146 }
147
148 /// Subscriptions that were in `previous` but are no longer desired, to be cancelled.
149 pub fn needs_unsubscribe(&self, previous: &SubscriptionSet) -> Vec<DesiredSubscription> {
150 previous
151 .subs()
152 .iter()
153 .filter(|sub| self.desired.find(sub.id).is_none())
154 .cloned()
155 .collect()
156 }
157}