photon_backend/models/
subscribe_opts.rs

1//! Options and handle for typed subscribe.
2
3use super::SubscriptionMode;
4
5/// Options specific to consumer-group subscriptions.
6///
7/// Used with [`SubscribeOpts::consumer_group`]. Checkpoints are owned by `group_id` × shard,
8/// not per-instance subscription names.
9#[derive(Debug, Clone)]
10pub struct GroupOpts {
11    /// Stable consumer group identifier (shared across replicas).
12    pub group_id: String,
13    /// Override virtual shard count (defaults to topic `#[photon::topic(shards = ...)]`).
14    pub shard_count: Option<u32>,
15    /// Fleet member id for lease coordination (static env assignment when unset).
16    pub instance_id: Option<String>,
17}
18
19impl GroupOpts {
20    /// Create group options with the given `group_id`.
21    pub fn new(group_id: impl Into<String>) -> Self {
22        Self {
23            group_id: group_id.into(),
24            shard_count: None,
25            instance_id: None,
26        }
27    }
28
29    /// Override virtual shard count for this subscription.
30    #[must_use]
31    pub const fn shards(mut self, count: u32) -> Self {
32        self.shard_count = Some(count);
33        self
34    }
35
36    /// Set fleet member instance id for lease-based assignment.
37    #[must_use]
38    pub fn instance_id(mut self, id: impl Into<String>) -> Self {
39        self.instance_id = Some(id.into());
40        self
41    }
42}
43
44/// Options for subscribing to a topic.
45///
46/// Pass to the typed API from [`topic`](https://docs.rs/uf-photon/latest/photon/attr.topic.html):
47/// `EventType::subscribe_on(&photon, opts)`.
48///
49/// Runnable: `keyed_topic`, `consumer_group`. Getting started:
50/// [publish and subscribe](https://docs.rs/uf-photon/latest/photon/#4-publish-and-subscribe).
51///
52/// # Examples
53///
54/// Ephemeral stream with partition filter (see `keyed_topic` example):
55///
56/// ```rust,ignore
57/// use futures::StreamExt;
58/// use photon::{SubscribeOpts, topic};
59///
60/// #[topic(name = "examples.orders", keyed_by = "customer_id")]
61/// struct OrderPlaced { customer_id: String, amount_cents: u64 }
62///
63/// # async fn demo(photon: &photon::Photon) -> photon::Result<()> {
64/// let opts = SubscribeOpts::default_ephemeral().topic_key_filter("alice");
65/// let mut stream = OrderPlaced::subscribe_on(photon, opts).await?;
66/// let _ = stream.next().await;
67/// # Ok(())
68/// # }
69/// ```
70///
71/// Durable broadcast by subscription name:
72///
73/// ```rust,no_run
74/// use photon_backend::SubscribeOpts;
75///
76/// let opts = SubscribeOpts::broadcast().subscription_name("my-worker");
77/// ```
78#[derive(Debug, Clone, Default)]
79pub struct SubscribeOpts {
80    /// Subscription name (required for durable broadcast mode).
81    pub subscription_name: Option<String>,
82    /// Filter by topic key (for keyed topics).
83    pub topic_key_filter: Option<String>,
84    /// Durable or ephemeral mode.
85    pub mode: SubscriptionMode,
86    /// Consumer group (load-balanced); mutually exclusive with durable subscription name.
87    pub consumer_group: Option<GroupOpts>,
88}
89
90impl SubscribeOpts {
91    /// Create default options (ephemeral, no key filter).
92    #[must_use]
93    pub const fn default_ephemeral() -> Self {
94        Self {
95            subscription_name: None,
96            topic_key_filter: None,
97            mode: SubscriptionMode::Ephemeral,
98            consumer_group: None,
99        }
100    }
101
102    /// Broadcast durable subscription (today's default).
103    #[must_use]
104    pub fn broadcast() -> Self {
105        Self {
106            mode: SubscriptionMode::Durable,
107            ..Self::default_ephemeral()
108        }
109    }
110
111    /// Load-balanced consumer group subscription.
112    #[must_use]
113    pub fn consumer_group(mut self, opts: GroupOpts) -> Self {
114        self.subscription_name = None;
115        self.topic_key_filter = None;
116        self.mode = SubscriptionMode::Durable;
117        self.consumer_group = Some(opts);
118        self
119    }
120
121    /// Set subscription name (for durable mode).
122    #[must_use]
123    pub fn subscription_name(mut self, name: impl Into<String>) -> Self {
124        self.subscription_name = Some(name.into());
125        self
126    }
127
128    /// Set topic key filter.
129    #[must_use]
130    pub fn topic_key_filter(mut self, key: impl Into<String>) -> Self {
131        self.topic_key_filter = Some(key.into());
132        self
133    }
134
135    /// Set mode.
136    #[must_use]
137    pub const fn mode(mut self, mode: SubscriptionMode) -> Self {
138        self.mode = mode;
139        self
140    }
141}
142
143/// Placeholder for future subscription lifecycle control (cancel, pause).
144///
145/// v0.1 returns a no-op handle; dropping the subscribe stream task is the supported way to
146/// stop an ephemeral subscription today.
147#[derive(Debug, Default)]
148pub struct SubscriptionHandle {
149    _private: (),
150}
151
152impl SubscriptionHandle {
153    /// Create a new handle (reserved for future cancel/pause APIs).
154    #[must_use]
155    pub const fn new() -> Self {
156        Self { _private: () }
157    }
158}
photon_backend/models/subscribe_opts.rs

photon_backend/models/
subscribe_opts.rs
