watermelon 0.4.4

High level actor based implementation NATS Core and NATS Jetstream client implementation
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
102

use std::{num::NonZero, time::Duration};

use jiff::Timestamp;
use serde::{Deserialize, Serialize};
use watermelon_proto::Subject;

use super::{compression, duration, nullable_datetime, nullable_number, opposite_bool};

/// A Jetstream stream
#[derive(Debug, Deserialize)]
pub struct Stream {
    pub config: StreamConfig,
    #[serde(rename = "created")]
    pub created_at: Timestamp,
    // TODO: `cluster`
}

/// The state of the stream
#[derive(Debug, Deserialize)]
pub struct StreamState {
    pub messages: u64,
    pub bytes: u64,
    pub first_sequence: u64,
    #[serde(with = "nullable_datetime", rename = "first_ts")]
    pub first_sequence_timestamp: Option<Timestamp>,
    pub last_sequence: u64,
    #[serde(with = "nullable_datetime", rename = "last_ts")]
    pub last_sequence_timestamp: Option<Timestamp>,
    pub consumer_count: u32,
}

/// A Jetstream stream configuration
#[derive(Debug, Serialize, Deserialize)]
#[expect(
    clippy::struct_excessive_bools,
    reason = "it is the actual config of a Jetstream"
)]
pub struct StreamConfig {
    pub name: String,
    pub subjects: Vec<Subject>,
    #[serde(with = "nullable_number")]
    pub max_consumers: Option<u32>,
    #[serde(with = "nullable_number", rename = "max_msgs")]
    pub max_messages: Option<u64>,
    #[serde(with = "nullable_number")]
    pub max_bytes: Option<u64>,
    #[serde(with = "duration")]
    pub max_age: Duration,
    #[serde(with = "nullable_number", rename = "max_msgs_per_subject")]
    pub max_messages_per_subject: Option<u64>,
    #[serde(with = "nullable_number", rename = "max_msg_size")]
    pub max_message_size: Option<u32>,
    #[serde(rename = "discard")]
    pub discard_policy: DiscardPolicy,
    pub storage: Storage,
    #[serde(rename = "num_replicas")]
    pub replicas: NonZero<u32>,
    #[serde(with = "duration")]
    pub duplicate_window: Duration,
    #[serde(with = "compression")]
    pub compression: Option<Compression>,
    pub allow_direct: bool,
    pub mirror_direct: bool,
    pub sealed: bool,
    #[serde(with = "opposite_bool", rename = "deny_delete")]
    pub allow_delete: bool,
    #[serde(with = "opposite_bool", rename = "deny_purge")]
    pub allow_purge: bool,
    pub allow_rollup_hdrs: bool,
    // TODO: `consumer_limits` https://github.com/nats-io/nats-server/blob/e25d973a8f389ce3aa415e4bcdfba1f7d0834f7f/server/stream.go#L99
}

/// A streams retention policy
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RetentionPolicy {
    Limits,
    Interest,
    WorkQueue,
}

/// A streams discard policy
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum DiscardPolicy {
    Old,
    New,
}

/// Whether the disk is stored on disk or in memory
#[derive(Debug, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Storage {
    File,
    Memory,
}

/// The compression algorithm used by a stream
#[derive(Debug)]
pub enum Compression {
    S2,
}