nakadi_types/subscription/
mod.rs

1//! Types for subscribing to an `EventType`
2use std::convert::AsRef;
3use std::fmt;
4use std::time::Duration;
5
6use chrono::{DateTime, Utc};
7use serde::{Deserialize, Deserializer, Serialize};
8use uuid::Uuid;
9
10use crate::misc::{AuthorizationAttribute, AuthorizationAttributes, OwningApplication};
11use crate::partition::{Cursor, CursorOffset, PartitionId};
12use crate::Error;
13
14mod subscription_input;
15pub use crate::event_type::EventTypeName;
16pub use subscription_input::*;
17
18/// Cursor of a subscription without a `CursorToken`.
19///
20/// This is a type alias of `EventTypeCursor` for people who want to use this long
21/// name which is used in the original Nakadi API.
22///
23/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionCursorWithoutToken)
24pub type SubscriptionCursorWithoutToken = EventTypeCursor;
25
26new_type! {
27    #[doc=r#"Id of subscription that was created.
28
29Is generated by Nakadi, should not be specified when creating
30subscription.
31
32See also [Nakadi Manual](https://nakadi.io/manual.html#definition_Subscription)"#]
33    #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
34    pub copy struct SubscriptionId(Uuid, env="SUBSCRIPTION_ID");
35}
36
37impl SubscriptionId {
38    pub fn random() -> Self {
39        Self(Uuid::new_v4())
40    }
41}
42
43new_type! {
44    #[doc=r#"The id of this stream generated by Nakadi.
45
46Must be used for committing events that were read by client from this stream.
47
48See also [Nakadi Manual](https://nakadi.io/manual.html#/subscriptions/subscription_id/events_get)"#]
49    #[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
50    pub copy struct StreamId(Uuid);
51}
52
53impl StreamId {
54    pub fn random() -> Self {
55        Self(Uuid::new_v4())
56    }
57}
58
59/// Something that has an event type and a partition
60///
61/// Must only return event types and partitions that belong together.
62pub trait EventTypePartitionLike {
63    fn event_type(&self) -> &EventTypeName;
64    fn partition(&self) -> &PartitionId;
65}
66
67/// Represents event-type/partition pair.
68///
69/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_EventTypePartition)
70#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Serialize, Deserialize)]
71pub struct EventTypePartition {
72    pub event_type: EventTypeName,
73    pub partition: PartitionId,
74}
75
76impl EventTypePartition {
77    pub fn new<E: Into<EventTypeName>, P: Into<PartitionId>>(event_type: E, partition: P) -> Self {
78        Self {
79            event_type: event_type.into(),
80            partition: partition.into(),
81        }
82    }
83
84    /// Split this instance into the `EventTypeName` and `PartitionId`.
85    pub fn split(self) -> (EventTypeName, PartitionId) {
86        (self.event_type, self.partition)
87    }
88
89    /// Get a reference to the `EventTypeName`
90    pub fn event_type(&self) -> &EventTypeName {
91        &self.event_type
92    }
93
94    /// Get a reference to the `PartitionId`
95    pub fn partition(&self) -> &PartitionId {
96        &self.partition
97    }
98}
99
100impl fmt::Display for EventTypePartition {
101    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102        write!(
103            f,
104            "[event_type={};partition={}]",
105            self.event_type, self.partition
106        )?;
107        Ok(())
108    }
109}
110
111impl EventTypePartitionLike for EventTypePartition {
112    fn event_type(&self) -> &EventTypeName {
113        &self.event_type
114    }
115
116    fn partition(&self) -> &PartitionId {
117        &self.partition
118    }
119}
120
121impl From<SubscriptionCursor> for EventTypePartition {
122    fn from(v: SubscriptionCursor) -> Self {
123        v.into_event_type_partition()
124    }
125}
126
127impl From<EventTypeCursor> for EventTypePartition {
128    fn from(v: EventTypeCursor) -> Self {
129        Self {
130            event_type: v.event_type,
131            partition: v.cursor.partition,
132        }
133    }
134}
135
136/// A collection of `EventTypeName`s
137///
138/// Used for e.g. creating subscriptions.
139#[derive(Debug, Clone, Default, Serialize, Deserialize)]
140pub struct EventTypeNames(Vec<EventTypeName>);
141
142impl EventTypeNames {
143    pub fn new<T: Into<Vec<EventTypeName>>>(event_types: T) -> Self {
144        Self(event_types.into())
145    }
146
147    pub fn event_type_name<T: Into<EventTypeName>>(mut self, v: T) -> Self {
148        self.0.push(v.into());
149        self
150    }
151
152    /// Return the inner `Vec` and consume self.
153    pub fn into_inner(self) -> Vec<EventTypeName> {
154        self.0
155    }
156
157    pub fn len(&self) -> usize {
158        self.0.len()
159    }
160
161    pub fn is_empty(&self) -> bool {
162        self.0.is_empty()
163    }
164
165    pub fn push<T: Into<EventTypeName>>(&mut self, event_type: T) {
166        self.0.push(event_type.into())
167    }
168}
169
170impl AsRef<[EventTypeName]> for EventTypeNames {
171    fn as_ref(&self) -> &[EventTypeName] {
172        &self.0
173    }
174}
175
176new_type! {
177/// The value describing the use case of this subscription.
178///
179/// In general that is an additional identifier used to differ subscriptions having the same
180/// owning_application and event_types.
181#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)]
182    pub struct ConsumerGroup(String, env="CONSUMER_GROUP");
183}
184
185/// The value describing the use case of this subscription.
186///
187/// In general that is an additional identifier used to differ subscriptions having the same
188/// owning_application and event_types.
189
190/// Status of one event-type within a context of subscription
191///
192/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStatus)
193#[derive(Debug, Clone, Serialize, Deserialize)]
194pub struct SubscriptionEventTypeStatus {
195    pub event_type: EventTypeName,
196    pub partitions: Vec<SubscriptionPartitionStatus>,
197}
198
199/// Status of one event-type within a context of subscription
200///
201/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStatus)
202#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
203pub struct SubscriptionPartitionStatus {
204    /// The partition id
205    pub partition: PartitionId,
206    /// The state of this partition in current subscription.
207    pub state: PartitionState,
208    /// The id of the stream that consumes data from this partition
209    pub stream_id: Option<StreamId>,
210    pub assignment_type: Option<PartitionAssignmentType>,
211}
212
213/// Assignment state of a partition within a context of subscription
214///
215/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStatus)
216#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
217pub enum PartitionState {
218    /// The partition is currently not assigned to any client
219    #[serde(rename = "unassigned")]
220    Unassigned,
221    /// The partition is currently reassigning from one client to another
222    #[serde(rename = "reassigned")]
223    Reassigned,
224    /// The partition is assigned to a client.
225    #[serde(rename = "assigned")]
226    Assigned,
227}
228
229/// Assignment type of a partition within a context of subscription
230///
231/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStatus)
232#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
233pub enum PartitionAssignmentType {
234    /// Partition can’t be transferred to another stream until the stream is closed
235    #[serde(rename = "direct")]
236    Direct,
237    /// Partition can be transferred to another stream in case of rebalance, or if another stream
238    /// requests to read from this partition.
239    #[serde(rename = "auto")]
240    Auto,
241}
242
243/// Authorization section of a Subscription
244///
245/// This section defines two access control lists: one for consuming events and
246/// committing cursors (‘readers’), and one for administering a subscription (‘admins’).
247/// Regardless of the values of the authorization properties,
248/// administrator accounts will always be authorized.
249///
250/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionAuthorization)
251#[derive(Debug, Default, Clone, PartialEq, Eq, Serialize, Deserialize)]
252pub struct SubscriptionAuthorization {
253    pub admins: AuthorizationAttributes,
254    pub readers: AuthorizationAttributes,
255}
256
257impl SubscriptionAuthorization {
258    pub fn new<A, R>(admins: A, readers: R) -> Self
259    where
260        A: Into<AuthorizationAttributes>,
261        R: Into<AuthorizationAttributes>,
262    {
263        Self {
264            admins: admins.into(),
265            readers: readers.into(),
266        }
267    }
268
269    pub fn admin<T: Into<AuthorizationAttribute>>(mut self, admin: T) -> Self {
270        self.admins.push(admin.into());
271        self
272    }
273    pub fn reader<T: Into<AuthorizationAttribute>>(mut self, reader: T) -> Self {
274        self.readers.push(reader.into());
275        self
276    }
277
278    pub fn add_admin<T: Into<AuthorizationAttribute>>(&mut self, admin: T) {
279        self.admins.push(admin.into())
280    }
281    pub fn add_reader<T: Into<AuthorizationAttribute>>(&mut self, reader: T) {
282        self.readers.push(reader.into())
283    }
284}
285
286/// Subscription is a high level consumption unit.
287///
288/// Subscriptions allow applications to easily scale the number of clients by managing
289/// consumed event offsets and distributing load between instances.
290/// The key properties that identify subscription are ‘owning_application’, ‘event_types’ and ‘consumer_group’.
291/// It’s not possible to have two different subscriptions with these properties being the same.
292///
293/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_Subscription)
294#[derive(Debug, Clone, Deserialize)]
295pub struct Subscription {
296    pub id: SubscriptionId,
297    pub owning_application: OwningApplication,
298    pub event_types: EventTypeNames,
299    #[serde(skip_serializing_if = "Option::is_none")]
300    pub consumer_group: Option<ConsumerGroup>,
301    pub created_at: DateTime<Utc>,
302    pub updated_at: DateTime<Utc>,
303}
304
305/// Cursor of a subscription without a `CursorToken`.
306///
307/// In the Nakadi API this type is named `SubscriptionCursorWithoutToken`. Since
308/// the original name is cumbersome it is renamed to `EventTypeCursor`
309/// There is an alias `SubscriptionCursorWithoutToken` for still using the
310/// original name.
311///
312/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionCursorWithoutToken)
313#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
314pub struct EventTypeCursor {
315    #[serde(flatten)]
316    pub cursor: Cursor,
317    /// The name of the event type this partition’s events belong to.
318    pub event_type: EventTypeName,
319}
320
321impl EventTypePartitionLike for EventTypeCursor {
322    fn event_type(&self) -> &EventTypeName {
323        &self.event_type
324    }
325
326    fn partition(&self) -> &PartitionId {
327        &self.cursor.partition
328    }
329}
330
331impl From<SubscriptionCursor> for EventTypeCursor {
332    fn from(c: SubscriptionCursor) -> Self {
333        c.into_event_type_cursor()
334    }
335}
336
337impl<'de> Deserialize<'de> for EventTypeCursor {
338    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
339    where
340        D: Deserializer<'de>,
341    {
342        #[derive(Deserialize)]
343        struct A {
344            #[serde(flatten)]
345            cursor: Cursor,
346            event_type: EventTypeName,
347        }
348
349        #[derive(Deserialize)]
350        struct B {
351            cursor: Cursor,
352            event_type: EventTypeName,
353        }
354
355        #[derive(Deserialize)]
356        #[serde(untagged)]
357        enum Variant {
358            A(A),
359            B(B),
360        }
361
362        let variant = Variant::deserialize(deserializer)?;
363
364        match variant {
365            Variant::A(x) => Ok(EventTypeCursor {
366                cursor: x.cursor,
367                event_type: x.event_type,
368            }),
369            Variant::B(x) => Ok(EventTypeCursor {
370                cursor: x.cursor,
371                event_type: x.event_type,
372            }),
373        }
374    }
375}
376
377/// An opaque value defined by the server.
378#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
379pub struct CursorToken(String);
380
381impl CursorToken {
382    pub fn new<T: Into<String>>(token: T) -> Self {
383        Self(token.into())
384    }
385}
386
387/// Cursor of a subscription with a `CursorToken` which is usually found within a
388/// `SubscriptionEventStreamBatch`.
389///
390/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionCursor)
391#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
392pub struct SubscriptionCursor {
393    #[serde(flatten)]
394    pub cursor: Cursor,
395    /// The name of the event type this partition’s events belong to.
396    pub event_type: EventTypeName,
397    /// An opaque value defined by the server.
398    pub cursor_token: CursorToken,
399}
400
401impl SubscriptionCursor {
402    pub fn into_event_type_cursor(self) -> EventTypeCursor {
403        EventTypeCursor {
404            cursor: self.cursor,
405            event_type: self.event_type,
406        }
407    }
408
409    /// Turns this into a `EventTypeCursor` that points to the begin of the subscription
410    pub fn into_event_type_cursor_begin(self) -> EventTypeCursor {
411        EventTypeCursor {
412            cursor: Cursor {
413                partition: self.cursor.partition,
414                offset: CursorOffset::Begin,
415            },
416            event_type: self.event_type,
417        }
418    }
419
420    pub fn into_event_type_partition(self) -> EventTypePartition {
421        EventTypePartition {
422            event_type: self.event_type,
423            partition: self.cursor.partition,
424        }
425    }
426
427    pub fn to_event_type_partition(&self) -> EventTypePartition {
428        EventTypePartition {
429            event_type: self.event_type.clone(),
430            partition: self.cursor.partition.clone(),
431        }
432    }
433}
434
435impl EventTypePartitionLike for SubscriptionCursor {
436    fn event_type(&self) -> &EventTypeName {
437        &self.event_type
438    }
439
440    fn partition(&self) -> &PartitionId {
441        &self.cursor.partition
442    }
443}
444
445/// The result of all cursor commits with the `SubscriptionCursor`s themselves.
446///
447/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_CursorCommitResult)
448#[derive(Debug, Default, Clone, Serialize, Deserialize)]
449pub struct CursorCommitResults {
450    #[serde(rename = "items")]
451    pub commit_results: Vec<CursorCommitResult>,
452}
453
454impl CursorCommitResults {
455    pub fn all_committed(&self) -> bool {
456        self.commit_results
457            .iter()
458            .all(|r| r.result == CommitResult::Committed)
459    }
460
461    pub fn iter_committed_cursors(&self) -> impl Iterator<Item = &SubscriptionCursor> {
462        self.commit_results
463            .iter()
464            .filter(|r| r.is_committed())
465            .map(|r| &r.cursor)
466    }
467
468    pub fn into_iter_committed_cursors(self) -> impl Iterator<Item = SubscriptionCursor> {
469        self.commit_results
470            .into_iter()
471            .filter(|r| r.is_committed())
472            .map(|r| r.cursor)
473    }
474
475    pub fn iter_outdated_cursors(&self) -> impl Iterator<Item = &SubscriptionCursor> {
476        self.commit_results
477            .iter()
478            .filter(|r| r.is_outdated())
479            .map(|r| &r.cursor)
480    }
481
482    pub fn into_iter_outdated_cursors(self) -> impl Iterator<Item = SubscriptionCursor> {
483        self.commit_results
484            .into_iter()
485            .filter(|r| r.is_outdated())
486            .map(|r| r.cursor)
487    }
488
489    pub fn into_inner(self) -> Vec<CursorCommitResult> {
490        self.commit_results
491    }
492}
493
494impl From<CursorCommitResults> for Vec<CursorCommitResult> {
495    fn from(v: CursorCommitResults) -> Self {
496        v.into_inner()
497    }
498}
499
500/// The result of single cursor commit with the `SubscriptionCursor` itself.
501///
502/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_CursorCommitResult)
503#[derive(Debug, Clone, Serialize, Deserialize)]
504pub struct CursorCommitResult {
505    pub cursor: SubscriptionCursor,
506    pub result: CommitResult,
507}
508
509impl CursorCommitResult {
510    pub fn is_committed(&self) -> bool {
511        self.result == CommitResult::Committed
512    }
513
514    pub fn is_outdated(&self) -> bool {
515        self.result == CommitResult::Outdated
516    }
517}
518
519/// The result of cursor commit.
520///
521/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_CursorCommitResult)
522#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
523#[serde(rename_all = "snake_case")]
524pub enum CommitResult {
525    /// Cursor was successfully committed
526    Committed,
527    /// There already was more recent (or the same) cursor committed,
528    /// so the current one was not committed as it is outdated
529    Outdated,
530}
531
532#[derive(Debug, Clone, Serialize, Deserialize)]
533pub struct SubscriptionStats {
534    #[serde(rename = "items")]
535    pub event_type_stats: Vec<SubscriptionEventTypeStats>,
536}
537
538impl SubscriptionStats {
539    pub fn unconsumed_events(&self) -> Option<usize> {
540        let mut is_sum = false;
541        let sum = self
542            .event_type_stats
543            .iter()
544            .filter_map(|set_stats| {
545                if set_stats.unconsumed_events().is_some() {
546                    is_sum = true;
547                }
548
549                set_stats.unconsumed_events()
550            })
551            .sum();
552
553        if is_sum {
554            Some(sum)
555        } else {
556            None
557        }
558    }
559
560    /// Returns the number of unconsumed events for this event type on the
561    /// given stream
562    pub fn unconsumed_stream_events(&self, for_stream: StreamId) -> Option<usize> {
563        let mut is_sum = false;
564        let sum = self
565            .event_type_stats
566            .iter()
567            .filter_map(|event_type_stats| {
568                if let Some(unconsumend) = event_type_stats.unconsumed_stream_events(for_stream) {
569                    is_sum = true;
570                    Some(unconsumend)
571                } else {
572                    None
573                }
574            })
575            .sum();
576
577        if is_sum {
578            Some(sum)
579        } else {
580            None
581        }
582    }
583
584    pub fn all_events_consumed(&self) -> bool {
585        self.unconsumed_events() == Some(0)
586    }
587}
588
589/// Statistics of one `EventType` within a context of subscription.
590///
591/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStats)
592#[derive(Debug, Clone, Serialize, Deserialize)]
593pub struct SubscriptionEventTypeStats {
594    pub event_type: EventTypeName,
595    #[serde(default)]
596    pub partitions: Vec<SubscriptionEventTypePartitionStats>,
597}
598
599impl SubscriptionEventTypeStats {
600    /// Returns the number of unconsumed events for this event type
601    pub fn unconsumed_events(&self) -> Option<usize> {
602        let mut is_sum = false;
603        let sum = self
604            .partitions
605            .iter()
606            .filter_map(|p| {
607                if p.unconsumed_events.is_some() {
608                    is_sum = true;
609                }
610
611                p.unconsumed_events
612            })
613            .sum();
614
615        if is_sum {
616            Some(sum)
617        } else {
618            None
619        }
620    }
621
622    /// Returns `true` if all events on all partitions on all event types
623    /// have been consumed.
624    pub fn all_events_consumed(&self) -> bool {
625        self.unconsumed_events() == Some(0)
626    }
627
628    /// Returns the number of unconsumed events for this event type on the
629    /// given stream
630    pub fn unconsumed_stream_events(&self, for_stream: StreamId) -> Option<usize> {
631        let mut is_sum = false;
632        let sum = self
633            .partitions
634            .iter()
635            .filter_map(|p| {
636                p.stream_id.and_then(|s| {
637                    if s == for_stream {
638                        if p.unconsumed_events.is_some() {
639                            is_sum = true;
640                        }
641                        p.unconsumed_events
642                    } else {
643                        None
644                    }
645                })
646            })
647            .sum();
648
649        if is_sum {
650            Some(sum)
651        } else {
652            None
653        }
654    }
655}
656
657/// Statistics of one partition for an `EventType` within a context of subscription.
658///
659/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStats)
660#[derive(Debug, Clone, Serialize, Deserialize)]
661pub struct SubscriptionEventTypePartitionStats {
662    partition: PartitionId,
663    /// The amount of events in this partition that are not yet consumed within this subscription.
664    /// The property may be absent at the moment when no events were yet consumed from the partition in this
665    /// subscription (In case of read_from is BEGIN or END)
666    ///
667    /// If the event type uses ‘compact’ cleanup policy - then the actual number of unconsumed events can be
668    /// lower than the one reported in this field.
669    #[serde(skip_serializing_if = "Option::is_none")]
670    unconsumed_events: Option<usize>,
671    /// Subscription consumer lag for this partition in seconds. Measured as the age of the oldest event of
672    /// this partition that is not yet consumed within this subscription.
673    #[serde(skip_serializing_if = "Option::is_none")]
674    consumer_lag_seconds: Option<u64>,
675    /// The id of the stream that consumes data from this partition
676    #[serde(deserialize_with = "crate::deserialize_empty_string_is_none")]
677    #[serde(skip_serializing_if = "Option::is_none")]
678    stream_id: Option<StreamId>,
679    #[serde(skip_serializing_if = "Option::is_none")]
680    /// * direct: partition can’t be transferred to another stream until the stream is closed;
681    /// * auto: partition can be transferred to another stream in case of rebalance, or if another stream
682    /// requests to read from this partition.
683    assignment_type: Option<String>,
684    /// The state of this partition in current subscription. Currently following values are possible:
685    ///
686    /// * unassigned: the partition is currently not assigned to any client;
687    /// * reassigning: the partition is currently reassigning from one client to another;
688    /// * assigned: the partition is assigned to a client.
689    state: SubscriptionPartitionState,
690}
691
692/// The state of this partition in current subscription.
693///
694/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStats)
695#[derive(Debug, Clone, Serialize, Deserialize)]
696#[serde(rename_all = "snake_case")]
697pub enum SubscriptionPartitionState {
698    /// The partition is assigned to a client
699    Assigned,
700    /// The partition is currently not assigned to any client
701    Unassigned,
702    /// The partition is currently reassigning from one client to another
703    Reassigning,
704}
705
706/// See also [Nakadi Manual](https://nakadi.io/manual.html#definition_SubscriptionEventTypeStats)
707#[derive(Debug, Clone, Serialize, Deserialize)]
708#[serde(rename_all = "snake_case")]
709pub enum SubscriptionPartitionAssignmentType {
710    /// Partition can’t be transferred to another stream until the stream is closed
711    Direct,
712    /// Partition can be transferred to another stream in case of rebalance, or if another stream
713    /// requests to read from this partition
714    Auto,
715}
716
717#[cfg(test)]
718mod test {
719    use super::*;
720    use crate::partition::CursorOffset;
721
722    use serde_json::{self, json};
723
724    #[test]
725    fn subscription_cursor_without_token() {
726        let json = json!({
727            "event_type": "the event",
728            "partition": "the partition",
729            "offset": "12345",
730        });
731
732        let sample = EventTypeCursor {
733            event_type: EventTypeName::new("the event"),
734            cursor: Cursor {
735                partition: PartitionId::new("the partition"),
736                offset: CursorOffset::new("12345"),
737            },
738        };
739
740        assert_eq!(
741            serde_json::to_value(sample.clone()).unwrap(),
742            json,
743            "serialize"
744        );
745        assert_eq!(
746            serde_json::from_value::<EventTypeCursor>(json).unwrap(),
747            sample,
748            "deserialize"
749        );
750    }
751
752    #[test]
753    fn subscription_cursor() {
754        let json = json!({
755            "event_type": "the event",
756            "partition": "the partition",
757            "offset": "12345",
758            "cursor_token": "abcdef",
759        });
760
761        let sample = SubscriptionCursor {
762            event_type: EventTypeName::new("the event"),
763            cursor_token: CursorToken::new("abcdef"),
764            cursor: Cursor {
765                partition: PartitionId::new("the partition"),
766                offset: CursorOffset::new("12345"),
767            },
768        };
769
770        assert_eq!(
771            serde_json::to_value(sample.clone()).unwrap(),
772            json,
773            "serialize"
774        );
775        assert_eq!(
776            serde_json::from_value::<SubscriptionCursor>(json).unwrap(),
777            sample,
778            "deserialize"
779        );
780    }
781}
782
783/// Parameters for starting a new stream on a subscription
784///
785/// See also [Nakadi Manual](https://nakadi.io/manual.html#/subscriptions/subscription_id/events_post)
786///
787/// # Environment
788///
789/// When initialized/updated from the environment the following environment variable
790/// are used which by are by default prefixed with "NAKADION_" or a custom prefix "<prefix>_":
791///
792/// * "STREAM_MAX_UNCOMMITTED_EVENTS"
793/// * "STREAM_BATCH_LIMIT"
794/// * "STREAM_LIMIT"
795/// * "STREAM_BATCH_FLUSH_TIMEOUT_SECS"
796/// * "STREAM_BATCH_TIMESPAN_SECS"
797/// * "STREAM_TIMEOUT_SECS"
798/// * "STREAM_COMMIT_TIMEOUT_SECS"
799#[derive(Debug, Default, Clone, Serialize, Deserialize)]
800pub struct StreamParameters {
801    /// List of partitions to read from in this stream. If absent or empty - then the partitions will be
802    /// automatically assigned by Nakadi.
803    #[serde(skip_serializing_if = "Vec::is_empty")]
804    pub partitions: Vec<EventTypePartition>,
805    /// The maximum number of uncommitted events that Nakadi will stream before pausing the stream. When in
806    /// paused state and commit comes - the stream will resume.
807    #[serde(skip_serializing_if = "Option::is_none")]
808    pub max_uncommitted_events: Option<StreamMaxUncommittedEvents>,
809    /// Maximum number of Events in each chunk (and therefore per partition) of the stream.
810    #[serde(skip_serializing_if = "Option::is_none")]
811    pub batch_limit: Option<StreamBatchLimit>,
812    /// Maximum number of Events in this stream (over all partitions being streamed in this
813    /// connection)
814    #[serde(skip_serializing_if = "Option::is_none")]
815    pub stream_limit: Option<StreamLimit>,
816    /// Maximum time in seconds to wait for the flushing of each chunk (per partition).
817    #[serde(skip_serializing_if = "Option::is_none")]
818    #[serde(rename = "batch_flush_timeout")]
819    pub batch_flush_timeout_secs: Option<StreamBatchFlushTimeoutSecs>,
820    /// Useful for batching events based on their `received_at` timestamp. For example, if `batch_timespan` is 5
821    /// seconds then Nakadi would flush a batch as soon as the difference in time between the first and the
822    /// last event in the batch exceeds 5 seconds. It waits for an event outside of the window to signal the
823    /// closure of a batch.
824    #[serde(skip_serializing_if = "Option::is_none")]
825    #[serde(rename = "batch_timespan")]
826    pub batch_timespan_secs: Option<StreamBatchTimespanSecs>,
827    /// Maximum time in seconds a stream will live before connection is closed by the server.
828    ///
829    /// If 0 or unspecified will stream for 1h ±10min.
830    #[serde(skip_serializing_if = "Option::is_none")]
831    #[serde(rename = "stream_timeout")]
832    pub stream_timeout_secs: Option<StreamTimeoutSecs>,
833    /// Maximum amount of seconds that Nakadi will be waiting for commit after sending a batch to a client.
834    /// In case if commit does not come within this timeout, Nakadi will initialize stream termination, no
835    /// new data will be sent. Partitions from this stream will be assigned to other streams.
836    /// Setting commit_timeout to 0 is equal to setting it to the maximum allowed value - 60 seconds.
837    #[serde(skip_serializing_if = "Option::is_none")]
838    #[serde(rename = "commit_timeout")]
839    pub commit_timeout_secs: Option<StreamCommitTimeoutSecs>,
840}
841
842impl StreamParameters {
843    /// Initializes all fields from environment variables prefixed with "NAKADION_"
844    pub fn from_env() -> Result<Self, Error> {
845        let mut me = Self::default();
846        me.fill_from_env()?;
847        Ok(me)
848    }
849
850    /// Initializes all fields from environment variables prefixed with "[prefix]_"
851    ///
852    /// The underscore is omitted if `prefix` is empty
853    pub fn from_env_prefixed<T: AsRef<str>>(prefix: T) -> Result<Self, Error> {
854        let mut me = Self::default();
855        me.fill_from_env_prefixed(prefix)?;
856        Ok(me)
857    }
858
859    /// Initializes all fields from environment variables without any prefix
860    pub fn from_env_type_names() -> Result<Self, Error> {
861        let mut me = Self::default();
862        me.fill_from_env_type_names()?;
863        Ok(me)
864    }
865
866    /// Updates all not yet set fields from environment variables prefixed with "NAKADION_"
867    pub fn fill_from_env(&mut self) -> Result<(), Error> {
868        self.fill_from_env_prefixed(crate::helpers::NAKADION_PREFIX)
869    }
870
871    /// Updates all not yet set fields from environment variables without any prefix
872    pub fn fill_from_env_type_names(&mut self) -> Result<(), Error> {
873        self.fill_from_env_prefixed("")
874    }
875
876    /// Updates all not yet set fields from environment variables prefixed with "[prefix]_"
877    ///
878    /// The underscore is omitted if `prefix` is empty
879    pub fn fill_from_env_prefixed<T: AsRef<str>>(&mut self, prefix: T) -> Result<(), Error> {
880        if self.max_uncommitted_events.is_none() {
881            self.max_uncommitted_events =
882                StreamMaxUncommittedEvents::try_from_env_prefixed(prefix.as_ref())?;
883        }
884        if self.batch_limit.is_none() {
885            self.batch_limit = StreamBatchLimit::try_from_env_prefixed(prefix.as_ref())?;
886        }
887        if self.stream_limit.is_none() {
888            self.stream_limit = StreamLimit::try_from_env_prefixed(prefix.as_ref())?;
889        }
890        if self.batch_flush_timeout_secs.is_none() {
891            self.batch_flush_timeout_secs =
892                StreamBatchFlushTimeoutSecs::try_from_env_prefixed(prefix.as_ref())?;
893        }
894        if self.batch_timespan_secs.is_none() {
895            self.batch_timespan_secs =
896                StreamBatchTimespanSecs::try_from_env_prefixed(prefix.as_ref())?;
897        }
898        if self.stream_timeout_secs.is_none() {
899            self.stream_timeout_secs = StreamTimeoutSecs::try_from_env_prefixed(prefix.as_ref())?;
900        }
901        if self.commit_timeout_secs.is_none() {
902            self.commit_timeout_secs =
903                StreamCommitTimeoutSecs::try_from_env_prefixed(prefix.as_ref())?;
904        }
905
906        Ok(())
907    }
908
909    pub fn apply_defaults(&mut self) {
910        if self.max_uncommitted_events.is_none() {
911            self.max_uncommitted_events = Some(StreamMaxUncommittedEvents::default());
912        }
913        if self.batch_limit.is_none() {
914            self.batch_limit = Some(StreamBatchLimit::default());
915        }
916        if self.stream_limit.is_none() {
917            self.stream_limit = Some(StreamLimit::default());
918        }
919        if self.batch_flush_timeout_secs.is_none() {
920            self.batch_flush_timeout_secs = Some(StreamBatchFlushTimeoutSecs::default());
921        }
922        if self.batch_timespan_secs.is_none() {
923            self.batch_timespan_secs = Some(StreamBatchTimespanSecs::default());
924        }
925        if self.stream_timeout_secs.is_none() {
926            self.stream_timeout_secs = Some(StreamTimeoutSecs::default());
927        }
928        if self.commit_timeout_secs.is_none() {
929            self.commit_timeout_secs = Some(StreamCommitTimeoutSecs::default());
930        }
931    }
932
933    /// List of partitions to read from in this stream. If absent or empty - then the partitions will be
934    /// automatically assigned by Nakadi.
935    pub fn partitions(mut self, partitions: Vec<EventTypePartition>) -> Self {
936        self.partitions = partitions;
937        self
938    }
939    /// The maximum number of uncommitted events that Nakadi will stream before pausing the stream.
940    ///
941    /// When in
942    /// paused state and commit comes - the stream will resume.
943    pub fn max_uncommitted_events<T: Into<StreamMaxUncommittedEvents>>(mut self, value: T) -> Self {
944        self.max_uncommitted_events = Some(value.into());
945        self
946    }
947    /// Maximum number of Events in each chunk (and therefore per partition) of the stream.
948    pub fn batch_limit<T: Into<StreamBatchLimit>>(mut self, value: T) -> Self {
949        self.batch_limit = Some(value.into());
950        self
951    }
952    /// Maximum number of Events in this stream (over all partitions being streamed in this
953    /// connection)
954    pub fn stream_limit<T: Into<StreamLimit>>(mut self, value: T) -> Self {
955        self.stream_limit = Some(value.into());
956        self
957    }
958    /// Maximum time in seconds to wait for the flushing of each chunk (per partition).
959    pub fn batch_flush_timeout_secs<T: Into<StreamBatchFlushTimeoutSecs>>(
960        mut self,
961        value: T,
962    ) -> Self {
963        self.batch_flush_timeout_secs = Some(value.into());
964        self
965    }
966    /// Useful for batching events based on their received_at timestamp.
967    ///
968    /// For example, if `batch_timespan` is 5
969    /// seconds then Nakadi would flush a batch as soon as the difference in time between the first and the
970    /// last event in the batch exceeds 5 seconds. It waits for an event outside of the window to signal the
971    /// closure of a batch.
972    pub fn batch_timespan_secs<T: Into<StreamBatchTimespanSecs>>(mut self, value: T) -> Self {
973        self.batch_timespan_secs = Some(value.into());
974        self
975    }
976    /// Maximum time in seconds a stream will live before connection is closed by the server..
977    pub fn stream_timeout_secs<T: Into<StreamTimeoutSecs>>(mut self, value: T) -> Self {
978        self.stream_timeout_secs = Some(value.into());
979        self
980    }
981    /// Maximum amount of seconds that Nakadi will be waiting for commit after sending a batch to a client.
982    ///
983    /// In case if commit does not come within this timeout, Nakadi will initialize stream termination, no
984    /// new data will be sent. Partitions from this stream will be assigned to other streams.
985    /// Setting commit_timeout to 0 is equal to setting it to the maximum allowed value - 60 seconds.
986    pub fn commit_timeout_secs<T: Into<StreamCommitTimeoutSecs>>(mut self, value: T) -> Self {
987        self.commit_timeout_secs = Some(value.into());
988        self
989    }
990
991    /// Returns the configured value or the Nakadi default
992    pub fn effective_commit_timeout_secs(&self) -> u32 {
993        self.commit_timeout_secs
994            .map(|s| s.into_inner())
995            .unwrap_or(60)
996    }
997
998    /// Returns the configured value or the Nakadi default
999    pub fn effective_max_uncommitted_events(&self) -> u32 {
1000        self.max_uncommitted_events
1001            .map(|s| s.into_inner())
1002            .unwrap_or(10)
1003    }
1004
1005    /// Returns the configured value or the Nakadi default
1006    pub fn effective_batch_limit(&self) -> u32 {
1007        self.batch_limit.map(|s| s.into_inner()).unwrap_or(1)
1008    }
1009}
1010
1011new_type! {
1012    #[doc="The maximum number of uncommitted events that Nakadi will stream before pausing the stream.\n"]
1013    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1014    pub copy struct StreamMaxUncommittedEvents(u32, env="STREAM_MAX_UNCOMMITTED_EVENTS");
1015}
1016impl Default for StreamMaxUncommittedEvents {
1017    fn default() -> Self {
1018        10.into()
1019    }
1020}
1021new_type! {
1022    #[doc="Maximum number of Events in each chunk (and therefore per partition) of the stream.\n"]
1023    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1024    pub copy struct StreamBatchLimit(u32, env="STREAM_BATCH_LIMIT");
1025}
1026impl Default for StreamBatchLimit {
1027    fn default() -> Self {
1028        1.into()
1029    }
1030}
1031new_type! {
1032    #[doc="Maximum number of Events in this stream \
1033    (over all partitions being streamed in this connection).\n\n\
1034    0 means infinite"]
1035    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1036    pub copy struct StreamLimit(u32, env="STREAM_LIMIT");
1037}
1038impl Default for StreamLimit {
1039    fn default() -> Self {
1040        0.into()
1041    }
1042}
1043new_type! {
1044    #[doc="Maximum time in seconds to wait for the flushing of each chunk (per partition).\n"]
1045    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1046    pub secs struct StreamBatchFlushTimeoutSecs(u32, env="STREAM_BATCH_FLUSH_TIMEOUT_SECS");
1047}
1048impl Default for StreamBatchFlushTimeoutSecs {
1049    fn default() -> Self {
1050        30.into()
1051    }
1052}
1053new_type! {
1054    #[doc="Useful for batching events based on their received_at timestamp.\n"]
1055    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1056    pub secs struct StreamBatchTimespanSecs(u32, env="STREAM_BATCH_TIMESPAN_SECS");
1057}
1058impl Default for StreamBatchTimespanSecs {
1059    fn default() -> Self {
1060        0.into()
1061    }
1062}
1063new_type! {
1064    #[doc="Maximum time in seconds a stream will live before connection is closed by the server.\n"]
1065    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1066    pub secs struct StreamTimeoutSecs(u32, env="STREAM_TIMEOUT_SECS");
1067}
1068impl Default for StreamTimeoutSecs {
1069    fn default() -> Self {
1070        0.into()
1071    }
1072}
1073new_type! {
1074    #[doc="Maximum amount of seconds that Nakadi will be waiting for commit after sending a batch to a client.\n"]
1075    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
1076    pub secs struct StreamCommitTimeoutSecs(u32, env="STREAM_COMMIT_TIMEOUT_SECS");
1077}
1078impl Default for StreamCommitTimeoutSecs {
1079    fn default() -> Self {
1080        60.into()
1081    }
1082}
nakadi_types/subscription/mod.rs

nakadi_types/subscription/
mod.rs
