#1636

Committed 25 Feb 2025 04:07AM UTC coverage: 64.48% (+0.9%) from 63.597%

Build # #1636

Build Type

push

Committed by

Commit Message

dep-update: bump aws-config from 1.5.16 to 1.5.17 (#772)

Bumps [aws-config](https://github.com/smithy-lang/smithy-rs) from 1.5.16 to 1.5.17.
- [Release notes](https://github.com/smithy-lang/smithy-rs/releases)
- [Changelog](https://github.com/smithy-lang/smithy-rs/blob/main/CHANGELOG.md)
- [Commits](https://github.com/smithy-lang/smithy-rs/commits)

---
updated-dependencies:
- dependency-name: aws-config
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Run Details

1661 of 2576 relevant lines covered (64.48%)

1.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.5

/server/src/delta_cache_manager.rs

use dashmap::DashMap;
use tokio::sync::broadcast;
use tracing::error;
use unleash_types::client_features::DeltaEvent;

use crate::delta_cache::DeltaCache;

#[derive(Debug, Clone)]
pub enum DeltaCacheUpdate {
    Full(String),     // environment with a newly inserted cache
    Update(String),   // environment with an updated delta cache
    Deletion(String), // environment removed
}

pub struct DeltaCacheManager {
    caches: DashMap<String, DeltaCache>,
    update_sender: broadcast::Sender<DeltaCacheUpdate>,
}

impl Default for DeltaCacheManager {
    fn default() -> Self {
        Self::new()
    }
}

impl DeltaCacheManager {
    pub fn new() -> Self {
        let (tx, _rx) = broadcast::channel::<DeltaCacheUpdate>(16);
        Self {
            caches: DashMap::new(),
            update_sender: tx,
        }
    }

    pub fn subscribe(&self) -> broadcast::Receiver<DeltaCacheUpdate> {
        self.update_sender.subscribe()
    }

    pub fn get(&self, env: &str) -> Option<DeltaCache> {
        self.caches.get(env).map(|entry| entry.value().clone())
    }

    pub fn insert_cache(&self, env: &str, cache: DeltaCache) {
        self.caches.insert(env.to_string(), cache);
        let _ = self
            .update_sender
            .send(DeltaCacheUpdate::Full(env.to_string()));
    }

    pub fn update_cache(&self, env: &str, events: &[DeltaEvent]) {
        if let Some(mut cache) = self.caches.get_mut(env) {
            cache.add_events(events);
            let result = self
                .update_sender
                .send(DeltaCacheUpdate::Update(env.to_string()));
            if let Err(e) = result {
                error!("Unexpected broadcast error: {:#?}", e);
            }
        }
    }

    pub fn remove_cache(&self, env: &str) {
        self.caches.remove(env);
        let _ = self
            .update_sender
            .send(DeltaCacheUpdate::Deletion(env.to_string()));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::delta_cache::{DeltaCache, DeltaHydrationEvent};
    use unleash_types::client_features::{ClientFeature, DeltaEvent, Segment};

    #[test]
    fn test_insert_and_update_delta_cache() {
        let hydration = DeltaHydrationEvent {
            event_id: 1,
            features: vec![ClientFeature {
                name: "feature1".to_string(),
                ..Default::default()
            }],
            segments: vec![Segment {
                id: 1,
                constraints: vec![],
            }],
        };
        let max_length = 5;
        let delta_cache = DeltaCache::new(hydration, max_length);
        let delta_cache_manager = DeltaCacheManager::new();
        let env = "test-env";

        let mut rx = delta_cache_manager.subscribe();

        delta_cache_manager.insert_cache(env, delta_cache);

        match rx.try_recv() {
            Ok(DeltaCacheUpdate::Full(e)) => assert_eq!(e, env),
            e => panic!("Expected Full update, got {:?}", e),
        }

        let update_event = DeltaEvent::FeatureUpdated {
            event_id: 2,
            feature: ClientFeature {
                name: "feature1_updated".to_string(),
                ..Default::default()
            },
        };

        delta_cache_manager.update_cache(env, &[update_event.clone()]);

        match rx.try_recv() {
            Ok(DeltaCacheUpdate::Update(e)) => assert_eq!(e, env),
            e => panic!("Expected Update update, got {:?}", e),
        }

        let cache = delta_cache_manager.get(env).expect("Cache should exist");
        let events = cache.get_events();
        assert_eq!(events.last().unwrap(), &update_event);
    }

    #[test]
    fn test_remove_delta_cache() {
        let hydration = DeltaHydrationEvent {
            event_id: 1,
            features: vec![ClientFeature {
                name: "feature-a".to_string(),
                ..Default::default()
            }],
            segments: vec![],
        };
        let delta_cache = DeltaCache::new(hydration, 3);
        let delta_cache_manager = DeltaCacheManager::new();
        let env = "remove-env";

        delta_cache_manager.insert_cache(env, delta_cache);
        let mut rx = delta_cache_manager.subscribe();
        let _ = rx.try_recv();

        delta_cache_manager.remove_cache(env);
        match rx.try_recv() {
            Ok(DeltaCacheUpdate::Deletion(e)) => assert_eq!(e, env),
            e => panic!("Expected Deletion update, got {:?}", e),
        }
        assert!(delta_cache_manager.get(env).is_none());
    }
}

1	use dashmap::DashMap;
2	use tokio::sync::broadcast;
3	use tracing::error;
4	use unleash_types::client_features::DeltaEvent;
5
6	use crate::delta_cache::DeltaCache;
7
8	#[derive(Debug, Clone)]
9	pub enum DeltaCacheUpdate {
10	Full(String), // environment with a newly inserted cache
11	Update(String), // environment with an updated delta cache
12	Deletion(String), // environment removed
13	}
14
15	pub struct DeltaCacheManager {
16	caches: DashMap<String, DeltaCache>,
17	update_sender: broadcast::Sender<DeltaCacheUpdate>,
18	}
19
20	impl Default for DeltaCacheManager {
21	fn default() -> Self {	×
22	Self::new()	×
23	}
24	}
25
26	impl DeltaCacheManager {
27	pub fn new() -> Self {	2✔
28	let (tx, _rx) = broadcast::channel::<DeltaCacheUpdate>(16);	2✔
29	Self {
30	caches: DashMap::new(),	3✔
31	update_sender: tx,
32	}
33	}
34
35	pub fn subscribe(&self) -> broadcast::Receiver<DeltaCacheUpdate> {	2✔
36	self.update_sender.subscribe()	2✔
37	}
38
39	pub fn get(&self, env: &str) -> Option<DeltaCache> {	2✔
40	self.caches.get(env).map(\|entry\| entry.value().clone())	6✔
41	}
42
43	pub fn insert_cache(&self, env: &str, cache: DeltaCache) {	2✔
44	self.caches.insert(env.to_string(), cache);	4✔
45	let _ = self	4✔
46	.update_sender
47	.send(DeltaCacheUpdate::Full(env.to_string()));	4✔
48	}
49
50	pub fn update_cache(&self, env: &str, events: &[DeltaEvent]) {	2✔
51	if let Some(mut cache) = self.caches.get_mut(env) {	2✔
52	cache.add_events(events);	4✔
53	let result = self	4✔
54	.update_sender
55	.send(DeltaCacheUpdate::Update(env.to_string()));	4✔
56	if let Err(e) = result {	2✔
57	error!("Unexpected broadcast error: {:#?}", e);	×
58	}
59	}
60	}
61
62	pub fn remove_cache(&self, env: &str) {	1✔
63	self.caches.remove(env);	1✔
64	let _ = self	2✔
65	.update_sender
66	.send(DeltaCacheUpdate::Deletion(env.to_string()));	2✔
67	}
68	}
69
70	#[cfg(test)]
71	mod tests {
72	use super::*;
73	use crate::delta_cache::{DeltaCache, DeltaHydrationEvent};
74	use unleash_types::client_features::{ClientFeature, DeltaEvent, Segment};
75
76	#[test]
77	fn test_insert_and_update_delta_cache() {
78	let hydration = DeltaHydrationEvent {
79	event_id: 1,
80	features: vec![ClientFeature {
81	name: "feature1".to_string(),
82	..Default::default()
83	}],
84	segments: vec![Segment {
85	id: 1,
86	constraints: vec![],
87	}],
88	};
89	let max_length = 5;
90	let delta_cache = DeltaCache::new(hydration, max_length);
91	let delta_cache_manager = DeltaCacheManager::new();
92	let env = "test-env";
93
94	let mut rx = delta_cache_manager.subscribe();
95
96	delta_cache_manager.insert_cache(env, delta_cache);
97
98	match rx.try_recv() {
99	Ok(DeltaCacheUpdate::Full(e)) => assert_eq!(e, env),
100	e => panic!("Expected Full update, got {:?}", e),
101	}
102
103	let update_event = DeltaEvent::FeatureUpdated {
104	event_id: 2,
105	feature: ClientFeature {
106	name: "feature1_updated".to_string(),
107	..Default::default()
108	},
109	};
110
111	delta_cache_manager.update_cache(env, &[update_event.clone()]);
112
113	match rx.try_recv() {
114	Ok(DeltaCacheUpdate::Update(e)) => assert_eq!(e, env),
115	e => panic!("Expected Update update, got {:?}", e),
116	}
117
118	let cache = delta_cache_manager.get(env).expect("Cache should exist");
119	let events = cache.get_events();
120	assert_eq!(events.last().unwrap(), &update_event);
121	}
122
123	#[test]
124	fn test_remove_delta_cache() {
125	let hydration = DeltaHydrationEvent {
126	event_id: 1,
127	features: vec![ClientFeature {
128	name: "feature-a".to_string(),
129	..Default::default()
130	}],
131	segments: vec![],
132	};
133	let delta_cache = DeltaCache::new(hydration, 3);
134	let delta_cache_manager = DeltaCacheManager::new();
135	let env = "remove-env";
136
137	delta_cache_manager.insert_cache(env, delta_cache);
138	let mut rx = delta_cache_manager.subscribe();
139	let _ = rx.try_recv();
140
141	delta_cache_manager.remove_cache(env);
142	match rx.try_recv() {
143	Ok(DeltaCacheUpdate::Deletion(e)) => assert_eq!(e, env),
144	e => panic!("Expected Deletion update, got {:?}", e),
145	}
146	assert!(delta_cache_manager.get(env).is_none());
147	}
148	}

Unleash / unleash-edge / #1636

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous