wingfoil 4.0.1

graph based stream processing framework
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
101

//! ZMQ adapter — real-time pub/sub messaging using ØMQ sockets with optional
//! etcd-based service discovery.
//!
//! Provides two graph primitives:
//!
//! - [`zmq_sub`] — subscriber that connects to a ZMQ PUB socket
//! - [`ZeroMqPub::zmq_pub`] — publisher that binds a ZMQ PUB socket
//!
//! # Setup
//!
//! ZMQ is peer-to-peer — no broker process is required. The `zmq` feature
//! bundles `libzmq` at build time, so no system installation is needed.
//!
//! Enable the feature in `Cargo.toml`:
//!
//! ```toml
//! wingfoil = { version = "...", features = ["zmq"] }
//! ```
//!
//! # Direct pub/sub
//!
//! ```ignore
//! use std::time::Duration;
//! use wingfoil::adapters::zmq::{ZeroMqPub, zmq_sub};
//! use wingfoil::*;
//!
//! // Publisher — binds on 127.0.0.1:5556
//! ticker(Duration::from_millis(100))
//!     .count()
//!     .zmq_pub(5556, ())
//!     .run(RunMode::RealTime, RunFor::Forever)
//!     .unwrap();
//!
//! // Subscriber — direct address
//! let (data, status) = zmq_sub::<Vec<u8>>("tcp://localhost:5556")?;
//! data.for_each(|burst, _| {
//!     for msg in burst { println!("{msg:?}"); }
//! })
//! .run(RunMode::RealTime, RunFor::Forever)
//! .unwrap();
//! # Ok::<(), anyhow::Error>(())
//! ```
//!
//! # etcd-based discovery
//!
//! ```ignore
//! use std::time::Duration;
//! use wingfoil::adapters::zmq::{EtcdRegistry, ZeroMqPub, zmq_sub};
//! use wingfoil::*;
//!
//! // Publisher registers its address in etcd under "quotes".
//! std::thread::spawn(move || {
//!     ticker(Duration::from_millis(100))
//!         .count()
//!         .zmq_pub(5556, ("quotes", EtcdRegistry::new("http://etcd:2379")))
//!         .run(RunMode::RealTime, RunFor::Forever)
//!         .unwrap();
//! });
//!
//! // Subscriber looks up the publisher address from etcd.
//! let (data, _status) = zmq_sub::<u64>(("quotes", EtcdRegistry::new("http://etcd:2379")))?;
//! data.for_each(|burst, _| println!("{burst:?}"))
//!     .run(RunMode::RealTime, RunFor::Forever)
//!     .unwrap();
//! # Ok::<(), anyhow::Error>(())
//! ```

mod read;
pub mod registry;
mod write;

#[cfg(all(test, feature = "zmq-integration-test"))]
mod integration_tests;

pub use read::*;
pub use registry::{ZmqHandle, ZmqPubRegistration, ZmqRegistry, ZmqSubConfig};
pub use write::*;

#[cfg(feature = "etcd")]
pub use registry::EtcdRegistry;

/// Connection status reported by the subscriber socket monitor.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum ZmqStatus {
    Connected,
    #[default]
    Disconnected,
}

/// Internal per-message event wrapping either data or a status change.
#[derive(Debug, Clone)]
pub enum ZmqEvent<T> {
    Data(T),
    Status(ZmqStatus),
}

impl<T: Default> Default for ZmqEvent<T> {
    fn default() -> Self {
        ZmqEvent::Data(T::default())
    }
}