ruststream_zeromq/endpoint.rs
1//! [`ZmqEndpoint`]: an address plus the side of it this process takes, as a type.
2//!
3//! Two processes talk directly, so which side listens is a deployment decision, not a property
4//! of the transport; the endpoint states it. The side is a type parameter rather than a
5//! field because it decides what a subscription can promise about a retry copy, and that is
6//! settled where the mount site is compiled.
7
8use std::marker::PhantomData;
9
10use ruststream::ServerSpec;
11
12#[cfg(feature = "asyncapi")]
13use crate::bindings;
14use crate::error::ZmqError;
15
16/// The ZMTP version the underlying implementation greets a peer with.
17///
18/// The [`zeromq`](https://docs.rs/zeromq) crate sends `3.0` in every greeting and negotiates no
19/// other, so this is a fact of the client rather than a configured value; the generated document
20/// reports it as the server's protocol version.
21pub(crate) const ZMTP_VERSION: &str = "3.0";
22
23pub(crate) use sealed::Side;
24
25mod sealed {
26 /// The side of an endpoint, as the machinery the three patterns share reads it.
27 ///
28 /// It is the [`EndpointRole`](super::EndpointRole) a public type carries, read off that type
29 /// once, when the endpoint is built: a socket attaches per side, and the attach code is
30 /// shared by every pattern. Public inside this private module, so it can ride the sealed
31 /// trait without being nameable outside the crate.
32 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
33 pub enum Side {
34 Bind,
35 Connect,
36 }
37
38 /// Keeps the sides at the two a socket can take, and carries the side the shared machinery
39 /// reads.
40 pub trait Sealed {
41 const SIDE: Side;
42 }
43}
44
45/// Which side of an endpoint this process takes: [`Bind`] or [`Connect`].
46///
47/// The endpoint constructors fix it - [`ZmqEndpoint::bind`] and [`ZmqEndpoint::connect`] - and
48/// every broker built on the endpoint carries it as its `Role` parameter, so `ZmqQueue<Connect>`
49/// is a queue this process dials. The side decides where a retry copy of a one-way subscription
50/// can go: a subscription that binds takes a copy sent to its own listener, while one that dials
51/// reads from a peer that only sends, so its mount site names where the copies go.
52///
53/// Sealed: the two sides are the whole set.
54///
55/// # Examples
56///
57/// A service that builds every queue with the same settings, on whichever side the endpoint
58/// takes:
59///
60/// ```
61/// use ruststream_zeromq::EndpointRole;
62/// use ruststream_zeromq::queue::prelude::*;
63/// use serde::{Deserialize, Serialize};
64///
65/// #[derive(Deserialize)]
66/// struct Job {
67/// id: u64,
68/// }
69///
70/// #[derive(Serialize, Outgoing)]
71/// #[outgoing(name = "results")]
72/// struct Done {
73/// id: u64,
74/// }
75///
76/// #[subscriber("jobs", reply)]
77/// async fn work(job: &Job) -> Done {
78/// Done { id: job.id }
79/// }
80///
81/// fn queue<Role: EndpointRole>(endpoint: ZmqEndpoint<Role>) -> ZmqQueue<Role> {
82/// ZmqQueue::new(endpoint).read_ahead(nonzero!(64_usize))
83/// }
84///
85/// #[ruststream::app]
86/// fn app() -> impl App {
87/// let results = queue(ZmqEndpoint::connect("tcp://sink:5556")).bindable();
88/// let to_results = results.bind(Publish);
89/// RustStream::new(AppInfo::new("worker", "0.1.0"))
90/// .with_broker(queue(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
91/// b.include(work).out_reply(to_results);
92/// })
93/// .register_broker(results)
94/// }
95/// ```
96pub trait EndpointRole: sealed::Sealed + Send + Sync + 'static {}
97
98/// This process listens on the endpoint: the side [`ZmqEndpoint::bind`] takes.
99///
100/// # Examples
101///
102/// Routes for a queue this process binds retry on that queue and name no destination, because
103/// the listener takes the copies:
104///
105/// ```
106/// use std::time::Duration;
107///
108/// use ruststream_zeromq::Bind;
109/// use ruststream_zeromq::queue::prelude::*;
110/// use serde::Deserialize;
111///
112/// #[derive(Deserialize)]
113/// struct Job {
114/// id: u64,
115/// }
116///
117/// #[subscriber("jobs")]
118/// async fn work(job: &Job) -> HandlerOutcome {
119/// if job.id == 0 {
120/// return HandlerOutcome::retry_after(Duration::from_secs(30));
121/// }
122/// HandlerOutcome::ack()
123/// }
124///
125/// fn routes() -> impl RouterDef<ZmqQueue<Bind>> {
126/// Router::<ZmqQueue<Bind>>::new()
127/// .include(work)
128/// .max_attempts(nonzero!(5u32))
129/// .out_retry(Publish)
130/// .build()
131/// }
132///
133/// #[ruststream::app]
134/// fn app() -> impl App {
135/// RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
136/// ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
137/// |b| {
138/// b.include_router(routes());
139/// },
140/// )
141/// }
142/// ```
143#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
144pub struct Bind;
145
146/// This process dials the endpoint: the side [`ZmqEndpoint::connect`] takes.
147///
148/// # Examples
149///
150/// A test of a worker that pushes its results to a collector it dials addresses that queue by its
151/// side:
152///
153/// ```
154/// # #[cfg(feature = "testing")]
155/// # mod demo {
156/// use std::error::Error;
157///
158/// use ruststream::testing::TestApp;
159/// use ruststream_zeromq::Connect;
160/// use ruststream_zeromq::queue::prelude::*;
161/// use serde::{Deserialize, Serialize};
162///
163/// #[derive(Deserialize, Serialize, Outgoing)]
164/// struct Job {
165/// id: u64,
166/// }
167///
168/// #[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
169/// #[outgoing(name = "results")]
170/// struct Done {
171/// id: u64,
172/// }
173///
174/// #[subscriber("jobs", reply)]
175/// async fn work(job: &Job) -> Done {
176/// Done { id: job.id }
177/// }
178///
179/// pub fn app() -> RustStream {
180/// let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
181/// let results = collector.bind(Publish);
182/// RustStream::new(AppInfo::new("worker", "0.1.0"))
183/// .with_broker(ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
184/// b.include(work).out_reply(results);
185/// })
186/// .register_broker(collector)
187/// }
188///
189/// pub async fn the_result_reaches_the_collector() -> Result<(), Box<dyn Error>> {
190/// let tb = TestApp::start(app()).await?;
191///
192/// tb.broker::<ZmqQueue>()
193/// .message(&Job { id: 7 })
194/// .to("jobs")
195/// .publish()
196/// .await?;
197///
198/// tb.broker::<ZmqQueue<Connect>>()
199/// .published::<Done>("results")
200/// .assert_called_once()
201/// .with(&Done { id: 7 });
202/// tb.shutdown().await?;
203/// Ok(())
204/// }
205/// # }
206/// # #[cfg(feature = "testing")]
207/// # use std::error::Error;
208/// # #[cfg(feature = "testing")]
209/// # #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
210/// # async fn main() -> Result<(), Box<dyn Error>> {
211/// # demo::the_result_reaches_the_collector().await
212/// # }
213/// # #[cfg(not(feature = "testing"))]
214/// # fn main() {}
215/// ```
216#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
217pub struct Connect;
218
219impl Side {
220 /// The side a public `Role` names.
221 pub(crate) const fn of<Role: EndpointRole>() -> Self {
222 <Role as sealed::Sealed>::SIDE
223 }
224}
225
226impl sealed::Sealed for Bind {
227 const SIDE: Side = Side::Bind;
228}
229
230impl sealed::Sealed for Connect {
231 const SIDE: Side = Side::Connect;
232}
233
234impl EndpointRole for Bind {}
235
236impl EndpointRole for Connect {}
237
238/// An address (`tcp://...` or `ipc://...`) and the side of it this process takes, fixed by the
239/// constructor and carried in the type.
240///
241/// # Examples
242///
243/// A worker that listens for jobs from a producer on the same host and pushes its results to a
244/// collector it dials over the network:
245///
246/// ```
247/// use ruststream_zeromq::queue::prelude::*;
248/// use serde::{Deserialize, Serialize};
249///
250/// #[derive(Deserialize)]
251/// struct Job {
252/// id: u64,
253/// }
254///
255/// #[derive(Serialize, Outgoing)]
256/// #[outgoing(name = "results")]
257/// struct Done {
258/// id: u64,
259/// }
260///
261/// #[subscriber("jobs", reply)]
262/// async fn work(job: &Job) -> Done {
263/// Done { id: job.id }
264/// }
265///
266/// #[ruststream::app]
267/// fn app() -> impl App {
268/// let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
269/// let results = collector.bind(Publish);
270/// RustStream::new(AppInfo::new("worker", "0.1.0"))
271/// .with_broker(ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")), |b| {
272/// b.include(work).out_reply(results);
273/// })
274/// .register_broker(collector)
275/// }
276/// ```
277#[derive(Debug, Clone, PartialEq, Eq)]
278#[must_use]
279pub struct ZmqEndpoint<Role = Bind> {
280 inner: Endpoint,
281 role: PhantomData<Role>,
282}
283
284impl ZmqEndpoint<Bind> {
285 /// This process listens on `address`.
286 ///
287 /// # Examples
288 ///
289 /// A worker that listens on a local socket for the producers on its host:
290 ///
291 /// ```
292 /// use ruststream_zeromq::queue::prelude::*;
293 /// use serde::Deserialize;
294 ///
295 /// #[derive(Deserialize)]
296 /// struct Job {
297 /// id: u64,
298 /// }
299 ///
300 /// #[subscriber("jobs")]
301 /// async fn work(job: &Job) -> HandlerOutcome {
302 /// tracing::info!(job.id, "working on the job");
303 /// HandlerOutcome::ack()
304 /// }
305 ///
306 /// #[ruststream::app]
307 /// fn app() -> impl App {
308 /// RustStream::new(AppInfo::new("worker", "0.1.0")).with_broker(
309 /// ZmqQueue::new(ZmqEndpoint::bind("ipc:///tmp/jobs")),
310 /// |b| {
311 /// b.include(work);
312 /// },
313 /// )
314 /// }
315 /// ```
316 pub fn bind(address: impl Into<String>) -> Self {
317 Self::on(address.into())
318 }
319}
320
321impl ZmqEndpoint<Connect> {
322 /// This process dials out to `address`.
323 ///
324 /// # Examples
325 ///
326 /// A worker that pushes its results to the collector it dials:
327 ///
328 /// ```
329 /// use ruststream_zeromq::queue::prelude::*;
330 /// use serde::{Deserialize, Serialize};
331 ///
332 /// #[derive(Deserialize)]
333 /// struct Job {
334 /// id: u64,
335 /// }
336 ///
337 /// #[derive(Serialize, Outgoing)]
338 /// #[outgoing(name = "results")]
339 /// struct Done {
340 /// id: u64,
341 /// }
342 ///
343 /// #[subscriber("jobs", reply)]
344 /// async fn work(job: &Job) -> Done {
345 /// Done { id: job.id }
346 /// }
347 ///
348 /// #[ruststream::app]
349 /// fn app() -> impl App {
350 /// let collector = ZmqQueue::new(ZmqEndpoint::connect("tcp://collector:5560")).bindable();
351 /// let results = collector.bind(Publish);
352 /// RustStream::new(AppInfo::new("worker", "0.1.0"))
353 /// .with_broker(ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")), |b| {
354 /// b.include(work).out_reply(results);
355 /// })
356 /// .register_broker(collector)
357 /// }
358 /// ```
359 pub fn connect(address: impl Into<String>) -> Self {
360 Self::on(address.into())
361 }
362}
363
364impl<Role: EndpointRole> ZmqEndpoint<Role> {
365 fn on(address: String) -> Self {
366 Self {
367 inner: Endpoint {
368 address,
369 side: Side::of::<Role>(),
370 },
371 role: PhantomData,
372 }
373 }
374}
375
376impl<Role> ZmqEndpoint<Role> {
377 /// The address string.
378 ///
379 /// # Examples
380 ///
381 /// A worker that logs where it listens before it starts:
382 ///
383 /// ```
384 /// use ruststream_zeromq::queue::prelude::*;
385 /// use serde::Deserialize;
386 ///
387 /// #[derive(Deserialize)]
388 /// struct Job {
389 /// id: u64,
390 /// }
391 ///
392 /// #[subscriber("jobs")]
393 /// async fn work(job: &Job) -> HandlerOutcome {
394 /// tracing::info!(job.id, "working on the job");
395 /// HandlerOutcome::ack()
396 /// }
397 ///
398 /// #[ruststream::app]
399 /// fn app() -> impl App {
400 /// let endpoint = ZmqEndpoint::bind("tcp://0.0.0.0:5555");
401 /// tracing::info!(address = endpoint.address(), "worker listening");
402 /// RustStream::new(AppInfo::new("worker", "0.1.0"))
403 /// .with_broker(ZmqQueue::new(endpoint), |b| {
404 /// b.include(work);
405 /// })
406 /// }
407 /// ```
408 #[must_use]
409 pub fn address(&self) -> &str {
410 &self.inner.address
411 }
412
413 /// The endpoint the shared machinery holds, its side read off the type.
414 pub(crate) fn into_inner(self) -> Endpoint {
415 self.inner
416 }
417}
418
419/// An endpoint with its side read off the public type: what the three patterns' shared
420/// machinery holds, so one attach routine serves every broker whatever its `Role`.
421#[derive(Debug, Clone, PartialEq, Eq)]
422pub(crate) struct Endpoint {
423 address: String,
424 side: Side,
425}
426
427impl Endpoint {
428 /// The address string.
429 pub(crate) fn address(&self) -> &str {
430 &self.address
431 }
432
433 /// The side of the endpoint this process takes.
434 pub(crate) const fn side(&self) -> Side {
435 self.side
436 }
437
438 /// The address a client connects to, without the transport scheme: a host and optional port
439 /// on `tcp://`, the socket path on `ipc://`. That is the shape the generated document
440 /// specifies for a server's host.
441 ///
442 /// The network form is the framework's own URL rule ([`ServerSpec::host_from_url`]), which
443 /// drops the scheme and any userinfo an operator wrote into the endpoint, so a published
444 /// document carries no credential. `ZeroMQ` addresses none of its own that way - its CURVE
445 /// keys are socket options - so this guards a shape the transport does not use.
446 ///
447 /// An `ipc` address names a filesystem path rather than an authority, so it is kept whole: the
448 /// URL rule would cut it at its first separator, and `@` is an ordinary character in a path.
449 pub(crate) fn host(&self) -> String {
450 match self.address.split_once("://") {
451 Some(("ipc", path)) => path.to_owned(),
452 _ => ServerSpec::host_from_url(&self.address),
453 }
454 }
455
456 /// How every pattern of this crate describes itself: the coordinate a client connects to, the
457 /// protocol, the ZMTP version the implementation greets with, and this crate's own binding.
458 ///
459 /// One place rather than three, because the answer does not depend on the socket pair: what a
460 /// pattern adds of its own goes on its channels, through its publish policy.
461 pub(crate) fn server_spec(&self) -> ServerSpec {
462 let spec = ServerSpec::new(self.host(), "zeromq").protocol_version(ZMTP_VERSION);
463 #[cfg(feature = "asyncapi")]
464 let spec = spec.bindings(bindings::server(self));
465 spec
466 }
467
468 /// Rejects endpoints the implementation cannot serve, before any I/O.
469 pub(crate) fn validate(&self) -> Result<(), ZmqError> {
470 if self.address.starts_with("tcp://") || self.address.starts_with("ipc://") {
471 Ok(())
472 } else {
473 Err(ZmqError::Invalid(format!(
474 "'{}' must use the tcp:// or ipc:// transport",
475 self.address
476 )))
477 }
478 }
479}
480
481#[cfg(test)]
482mod tests {
483 use ruststream::DescribeServer;
484
485 use super::*;
486 use crate::{ZmqFanout, ZmqQueue, ZmqRpc};
487
488 #[test]
489 fn unsupported_transports_are_rejected_before_io() {
490 assert!(
491 ZmqEndpoint::bind("inproc://x")
492 .into_inner()
493 .validate()
494 .is_err()
495 );
496 assert!(
497 ZmqEndpoint::connect("udp://x:1")
498 .into_inner()
499 .validate()
500 .is_err()
501 );
502 assert!(
503 ZmqEndpoint::bind("tcp://0.0.0.0:5555")
504 .into_inner()
505 .validate()
506 .is_ok()
507 );
508 assert!(
509 ZmqEndpoint::bind("ipc:///tmp/x")
510 .into_inner()
511 .validate()
512 .is_ok()
513 );
514 }
515
516 /// The refusal names what was invalid, an endpoint, and the address it refused.
517 #[test]
518 fn a_refused_transport_reads_as_an_invalid_endpoint() {
519 let err = ZmqEndpoint::bind("inproc://x")
520 .into_inner()
521 .validate()
522 .expect_err("inproc is refused");
523 assert_eq!(
524 err.to_string(),
525 concat!(
526 "invalid zeromq endpoint or mount: ",
527 "'inproc://x' must use the tcp:// or ipc:// transport"
528 )
529 );
530 }
531
532 /// The constructor is the only place a side is chosen, and the value the machinery reads is
533 /// the one the type names.
534 #[test]
535 fn the_side_the_machinery_reads_is_the_one_the_type_names() {
536 assert_eq!(
537 ZmqEndpoint::bind("tcp://0.0.0.0:5555").into_inner().side(),
538 Side::Bind
539 );
540 assert_eq!(
541 ZmqEndpoint::connect("tcp://ml:5555").into_inner().side(),
542 Side::Connect
543 );
544 }
545
546 #[test]
547 fn the_host_drops_the_scheme_on_both_transports() {
548 assert_eq!(
549 ZmqEndpoint::connect("tcp://broker:5555")
550 .into_inner()
551 .host(),
552 "broker:5555"
553 );
554 assert_eq!(
555 ZmqEndpoint::bind("tcp://0.0.0.0:0").into_inner().host(),
556 "0.0.0.0:0"
557 );
558 assert_eq!(
559 ZmqEndpoint::bind("ipc:///tmp/orders").into_inner().host(),
560 "/tmp/orders"
561 );
562 }
563
564 #[test]
565 fn the_host_drops_userinfo_on_the_last_at_sign() {
566 assert_eq!(
567 ZmqEndpoint::connect("tcp://user:pa@ss@broker:5555")
568 .into_inner()
569 .host(),
570 "broker:5555"
571 );
572 // A path is not an authority, so an '@' in it is part of the name, not a separator.
573 assert_eq!(
574 ZmqEndpoint::bind("ipc:///tmp/a@b").into_inner().host(),
575 "/tmp/a@b"
576 );
577 }
578
579 /// Every pattern describes itself the same way, and the generated document specifies a host,
580 /// not a URL.
581 #[test]
582 fn no_pattern_puts_a_scheme_in_the_described_host() {
583 for address in ["tcp://broker:5555", "ipc:///tmp/orders"] {
584 let specs = [
585 ZmqQueue::new(ZmqEndpoint::connect(address)).describe_server(),
586 ZmqFanout::new(ZmqEndpoint::connect(address)).describe_server(),
587 ZmqRpc::new(ZmqEndpoint::connect(address)).describe_server(),
588 ];
589 for spec in specs {
590 let host = spec.host.expect("a networked broker describes a host");
591 assert!(!host.contains("://"), "host carries a scheme: {host}");
592 assert_eq!(spec.protocol, "zeromq");
593 }
594 }
595 }
596}