kevy_replicate/replica_connect.rs
1//! How a replica opens its link: [`ConnectOptions`] and
2//! [`ReplicaClient::connect_with`].
3
4use std::net::ToSocketAddrs;
5use std::time::Duration;
6
7use crate::feed::FeedPosition;
8use crate::replica::ReplicaClient;
9use crate::replica_error::ReplicaError;
10#[cfg(feature = "secure")]
11use crate::replica_secure::ReplicaSecurity;
12
13/// Everything a replica presents when it connects, beyond the address.
14/// The defaults are a fresh replica: generation 0 (no continuity claim),
15/// offset 0, a 5 s connect timeout, a plaintext link.
16///
17/// ```
18/// use std::time::Duration;
19/// use kevy_replicate::feed::FeedPosition;
20/// use kevy_replicate::replica::ConnectOptions;
21///
22/// let opts = ConnectOptions::new("replica-a")
23/// .with_from(FeedPosition::new(7, 42))
24/// .with_timeout(Duration::from_secs(1));
25/// assert_eq!(opts.from, FeedPosition::new(7, 42));
26/// ```
27#[derive(Debug, Clone)]
28#[non_exhaustive]
29pub struct ConnectOptions {
30 /// The replica's identifier, operator-set; the primary keys its slot
31 /// by it.
32 ///
33 /// ```
34 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
35 /// use kevy_replicate::handshake::HandshakeReq;
36 /// use kevy_replicate::replica::{ConnectOptions, ReplicaClient};
37 ///
38 /// # let (addr, primary) = fake_primary(b"+ACK 1 0\r\n", Vec::new());
39 /// drop(ReplicaClient::connect_with(addr, &ConnectOptions::new("replica-a"))?);
40 /// // the primary sees the id in the handshake
41 /// let sent = primary.join().expect("primary thread");
42 /// let mut argv = kevy_resp::Argv::default();
43 /// kevy_resp::parse_command_into(&sent, &mut argv)?;
44 /// assert_eq!(HandshakeReq::parse(&argv)?.replica_id, "replica-a");
45 /// # Ok::<(), Box<dyn std::error::Error>>(())
46 /// ```
47 pub replica_id: String,
48 /// Where to resume: the feed generation this replica's data reflects
49 /// (`0` = unknown or fresh) and the offset within it. The primary
50 /// serves the offset's continuity only when the generations match;
51 /// otherwise it ships a snapshot.
52 ///
53 /// ```
54 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
55 /// use kevy_replicate::feed::FeedPosition;
56 /// use kevy_replicate::replica::{ConnectOptions, ReplicaClient};
57 ///
58 /// # let (addr, _primary) = fake_primary(b"+ACK 7 42\r\n", Vec::new());
59 /// let opts = ConnectOptions::new("replica-a").with_from(FeedPosition::new(7, 42));
60 /// let client = ReplicaClient::connect_with(addr, &opts)?;
61 /// assert_eq!(client.expected_offset(), opts.from.offset); // resumes where its data ends
62 /// # Ok::<(), kevy_replicate::replica::ReplicaError>(())
63 /// ```
64 pub from: FeedPosition,
65 /// How long the connect and the handshake may take.
66 ///
67 /// ```
68 /// use std::time::Duration;
69 /// use kevy_replicate::replica::{ConnectOptions, ReplicaClient, ReplicaError};
70 ///
71 /// // a listener that never answers the handshake
72 /// let silent = std::net::TcpListener::bind("127.0.0.1:0")?;
73 /// let opts = ConnectOptions::new("replica-a").with_timeout(Duration::from_millis(50));
74 /// let err = ReplicaClient::connect_with(silent.local_addr()?, &opts).unwrap_err();
75 /// assert!(matches!(err, ReplicaError::Io(_)));
76 /// # Ok::<(), std::io::Error>(())
77 /// ```
78 pub timeout: Duration,
79 /// `Some` to open a Noise IK link instead of a plaintext one.
80 ///
81 /// ```
82 /// use kevy_replicate::replica::{ConnectOptions, ReplicaSecurity};
83 ///
84 /// assert!(ConnectOptions::new("r1").security.is_none()); // plaintext by default
85 /// let sec = ReplicaSecurity::new(kevy_noise::Keypair::from_secret([2; 32]), [9; 32]);
86 /// let opts = ConnectOptions::new("r1").with_security(sec);
87 /// assert_eq!(opts.security.map(|s| s.primary_key), Some([9; 32]));
88 /// ```
89 #[cfg(feature = "secure")]
90 pub security: Option<ReplicaSecurity>,
91}
92
93impl ConnectOptions {
94 /// A fresh replica named `replica_id`, with the defaults above.
95 ///
96 /// ```
97 /// let opts = kevy_replicate::replica::ConnectOptions::new("r1");
98 /// assert_eq!(opts.from, kevy_replicate::feed::FeedPosition::default());
99 /// ```
100 pub fn new(replica_id: impl Into<String>) -> Self {
101 ConnectOptions {
102 replica_id: replica_id.into(),
103 from: FeedPosition::default(),
104 timeout: Duration::from_secs(5),
105 #[cfg(feature = "secure")]
106 security: None,
107 }
108 }
109
110 /// Set [`ConnectOptions::from`].
111 ///
112 /// ```
113 /// use kevy_replicate::feed::FeedPosition;
114 ///
115 /// let opts = kevy_replicate::replica::ConnectOptions::new("r1").with_from(FeedPosition::new(3, 9));
116 /// assert_eq!(opts.from, FeedPosition::new(3, 9));
117 /// ```
118 #[must_use]
119 pub fn with_from(mut self, from: FeedPosition) -> Self {
120 self.from = from;
121 self
122 }
123
124 /// Set [`ConnectOptions::timeout`].
125 ///
126 /// ```
127 /// use std::time::Duration;
128 /// let opts = kevy_replicate::replica::ConnectOptions::new("r1").with_timeout(Duration::from_secs(1));
129 /// assert_eq!(opts.timeout, Duration::from_secs(1));
130 /// ```
131 #[must_use]
132 pub fn with_timeout(mut self, timeout: Duration) -> Self {
133 self.timeout = timeout;
134 self
135 }
136
137 /// Open the link under Noise IK with `security`.
138 ///
139 /// ```
140 /// use kevy_replicate::replica::{ConnectOptions, ReplicaSecurity};
141 ///
142 /// let sec = ReplicaSecurity::new(kevy_noise::Keypair::from_secret([2; 32]), [9; 32]);
143 /// let opts = ConnectOptions::new("r1").with_security(sec);
144 /// assert!(opts.security.is_some());
145 /// ```
146 #[cfg(feature = "secure")]
147 #[must_use]
148 pub fn with_security(mut self, security: ReplicaSecurity) -> Self {
149 self.security = Some(security);
150 self
151 }
152}
153
154impl ReplicaClient {
155 /// Connect to `addr` as `opts` describes, send `REPLICATE FROM
156 /// <generation> <from_offset> ID <replica_id>`, read the `+ACK <gen>
157 /// <offset>` reply, and return a ready-to-iterate client. A secure
158 /// link authenticates the primary by `opts.security`'s key before the
159 /// handshake, and everything after it is encrypted.
160 ///
161 /// ```
162 /// # mod doc { include!(concat!(env!("CARGO_MANIFEST_DIR"), "/tests/support/doc_primary.rs")); } use doc::*;
163 /// use kevy_replicate::replica::{ConnectOptions, ReplicaClient};
164 ///
165 /// # let (addr, _primary) = fake_primary(b"+ACK 7 42\r\n", Vec::new());
166 /// let opts = ConnectOptions::new("replica-a").with_from(kevy_replicate::feed::FeedPosition::new(7, 42));
167 /// let client = ReplicaClient::connect_with(addr, &opts)?;
168 /// assert_eq!(client.expected_offset(), 42);
169 /// # Ok::<(), kevy_replicate::replica::ReplicaError>(())
170 /// ```
171 pub fn connect_with<A: ToSocketAddrs>(
172 addr: A,
173 opts: &ConnectOptions,
174 ) -> Result<Self, ReplicaError> {
175 #[cfg(feature = "secure")]
176 if let Some(security) = &opts.security {
177 return Self::connect_noise(addr, opts, security);
178 }
179 Self::connect_plain(addr, opts)
180 }
181}