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
//! The rail's contract: what the network is asked, its counts, and how it
//! refuses.
use std::sync::atomic::{AtomicU64, Ordering};
use serde_json::Value;
/// How the network answered, or why it did not. The words of a refusal are
/// the server's; the ticket adds none.
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
pub enum RailError {
/// The server refused or could not answer; its words.
#[error("{0}")]
Refused(String),
/// A topology record larger than the configured limit.
#[error("a topology record over {limit} bytes was refused")]
RecordTooLarge {
/// The configured limit.
limit: usize,
},
/// No topology record arrived within the reply timeout.
#[error("no topology record arrived within {millis} ms")]
NoRecord {
/// The timeout waited.
millis: u64,
},
/// The record or the reply is not the JSON expected.
#[error("the network's answer could not be read: {message}")]
Unreadable {
/// The reader's words.
message: String,
},
}
/// What the ticket did on the network, counted (SDK-017 R5).
#[derive(Debug, Default)]
pub struct Counts {
/// Connections opened.
pub connections_opened: AtomicU64,
/// Connections closed.
pub connections_closed: AtomicU64,
/// Subscriptions opened on `console.topology`.
pub subscribes: AtomicU64,
/// Topology records received.
pub records: AtomicU64,
/// Requests sent on a subject.
pub requests: AtomicU64,
/// Replies received for them.
pub replies: AtomicU64,
}
impl Counts {
/// The six counts in order: opened, closed, subscribes, records,
/// requests, replies.
#[must_use]
pub fn read(&self) -> [u64; 6] {
[
self.connections_opened.load(Ordering::SeqCst),
self.connections_closed.load(Ordering::SeqCst),
self.subscribes.load(Ordering::SeqCst),
self.records.load(Ordering::SeqCst),
self.requests.load(Ordering::SeqCst),
self.replies.load(Ordering::SeqCst),
]
}
pub(crate) fn add(counter: &AtomicU64) {
counter.fetch_add(1, Ordering::SeqCst);
}
}
/// The network as the tools see it.
pub trait Rail: std::fmt::Debug {
/// The current topology record, read fresh: one subscription, the first
/// whole record, closed before returning.
///
/// # Errors
/// Returns the server's refusal, an oversize record or no record.
fn topology(&self) -> Result<Value, RailError>;
/// One request on `subject`; with `reply` the correlated reply's JSON,
/// without it `None` once sent.
///
/// # Errors
/// Returns the server's refusal in its words.
fn request(
&self,
subject: &str,
input: &Value,
reply: bool,
) -> Result<Option<Value>, RailError>;
/// The calling seat's own pass for `audience`, minted for a call that
/// names `pass_for`; never the pass the rail's connections present.
///
/// # Errors
/// Returns why no pass could be had, in words that carry no secret.
fn pass_for(&self, audience: &str) -> Result<String, RailError>;
/// Opens a long-lived subscription on a tap channel using the seat pass.
///
/// # Errors
/// Returns the server's refusal in its words.
fn tap(&self, channel: &str) -> Result<super::TapSubscription, RailError>;
/// The configured server address used to identify notification records.
fn server_address(&self) -> &str;
/// The counts so far.
fn counts(&self) -> &Counts;
/// The largest request body the rail sends.
fn request_bytes(&self) -> usize;
/// The most steps one flow may carry.
fn flow_steps(&self) -> usize;
}