photon-backend 0.1.0

Photon publish/subscribe backend implementations and storage adapters
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
103
104

//! Topic partition and subscription partition keys for reclaim.

use std::collections::HashSet;

use crate::handler_registry::HandlerRegistry;
use crate::retention::config::RetentionPolicy;
use crate::retention::hook::RetentionHook;

/// Storage partition `(topic, optional key)`.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct TopicPartition {
    /// Topic name.
    pub topic_name: String,
    /// Optional partition / shard key.
    pub topic_key: Option<String>,
}

impl TopicPartition {
    /// Create a partition key.
    pub fn new(topic_name: impl Into<String>, topic_key: Option<String>) -> Self {
        Self {
            topic_name: topic_name.into(),
            topic_key,
        }
    }
}

/// Durable subscription scoped to one transport partition.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct SubscriptionPartition {
    /// Durable subscription name.
    pub subscription_name: String,
    /// Topic name.
    pub topic_name: String,
    /// Optional partition / shard key.
    pub topic_key: Option<String>,
}

/// Collect known durable subscriptions for watermark checkpoint loads.
pub fn known_subscriptions(
    policy: &RetentionPolicy,
    hook: Option<&dyn RetentionHook>,
) -> Vec<SubscriptionPartition> {
    let mut seen = HashSet::new();
    let mut out = Vec::new();

    let registry = HandlerRegistry::auto_discover();
    for handler in registry.iter() {
        let entry = SubscriptionPartition {
            subscription_name: handler.subscription_name.to_string(),
            topic_name: handler.topic_name.to_string(),
            topic_key: None,
        };
        if seen.insert(entry.clone()) {
            out.push(entry);
        }
    }

    for entry in &policy.extra_subscriptions {
        if seen.insert(entry.clone()) {
            out.push(entry.clone());
        }
    }

    if let Some(h) = hook {
        for entry in h.extra_subscriptions() {
            if seen.insert(entry.clone()) {
                out.push(entry);
            }
        }
    }

    out
}

/// Subscriptions whose checkpoint applies to the given transport partition.
pub fn subscriptions_for_partition<'a>(
    subs: &'a [SubscriptionPartition],
    topic: &str,
    topic_key: Option<&str>,
) -> Vec<&'a SubscriptionPartition> {
    subs.iter()
        .filter(|s| {
            s.topic_name == topic
                && match (&s.topic_key, topic_key) {
                    (None, _) => true,
                    (Some(sk), Some(tk)) => sk == tk,
                    (Some(_), None) => false,
                }
        })
        .collect()
}

/// Merge partition sets from multiple sources.
pub fn merge_partitions(sources: impl IntoIterator<Item = TopicPartition>) -> Vec<TopicPartition> {
    let mut seen = HashSet::new();
    let mut out = Vec::new();
    for part in sources {
        if seen.insert(part.clone()) {
            out.push(part);
        }
    }
    out
}