refactor(memory): 将 store.rs 拆分为模块目录，仅结构搬移
- 将 src/memory/store.rs 单体文件拆分为模块根 + store/ 目录结构 - InMemoryStore（struct + impl + Default + 6 个内联测试）整体提取到 src/memory/store/in_memory.rs - store.rs 保留 MemoryStore trait 与 EvictionPolicy/EvictionConfig - 外部消费者导入路径 crate::memory::store::MemoryStore 不变
This commit is contained in:
徐涛
+4
-259
@@ -1,14 +1,14 @@
|
||||
//! MemoryStore 抽象接口与默认实现。
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
use crate::memory::error::MemoryError;
|
||||
use crate::memory::types::{MemoryFilter, MemoryItem};
|
||||
|
||||
pub mod in_memory;
|
||||
|
||||
pub use in_memory::InMemoryStore;
|
||||
|
||||
/// 底层记忆存储抽象接口。
|
||||
///
|
||||
/// 下游可实现此 trait 以对接持久化后端（JSON 文件、SQLite、Redis 等）。
|
||||
@@ -58,258 +58,3 @@ impl Default for EvictionConfig {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。
|
||||
pub struct InMemoryStore {
|
||||
items: Mutex<HashMap<String, MemoryItem>>,
|
||||
eviction: EvictionConfig,
|
||||
/// 自上次淘汰检查以来的写入次数。
|
||||
writes_since_check: Mutex<usize>,
|
||||
}
|
||||
|
||||
impl InMemoryStore {
|
||||
/// 创建一个无淘汰策略的 InMemoryStore。
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
items: Mutex::new(HashMap::new()),
|
||||
eviction: EvictionConfig::default(),
|
||||
writes_since_check: Mutex::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
/// 创建一个带淘汰配置的 InMemoryStore。
|
||||
pub fn with_eviction(eviction: EvictionConfig) -> Self {
|
||||
Self {
|
||||
items: Mutex::new(HashMap::new()),
|
||||
eviction,
|
||||
writes_since_check: Mutex::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn maybe_evict(&self) {
|
||||
// 不使用 .lock().await 跨点，先取计数判断是否需要淘汰
|
||||
let should_check = {
|
||||
let mut counter = self.writes_since_check.lock().unwrap();
|
||||
*counter += 1;
|
||||
if *counter >= self.eviction.check_interval {
|
||||
*counter = 0;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
};
|
||||
if !should_check {
|
||||
return;
|
||||
}
|
||||
|
||||
let policy = self.eviction.policy.clone();
|
||||
match policy {
|
||||
EvictionPolicy::None => {}
|
||||
EvictionPolicy::Ttl { ttl_secs } => {
|
||||
let cutoff = OffsetDateTime::now_utc() - time::Duration::seconds(ttl_secs as i64);
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.retain(|_, v| v.created_at > cutoff);
|
||||
}
|
||||
EvictionPolicy::Capacity { max_items } => {
|
||||
let mut items = self.items.lock().unwrap();
|
||||
if items.len() > max_items {
|
||||
let mut vec: Vec<_> = items.drain().collect();
|
||||
// O(n) 部分排序：保留 created_at 最大的 max_items 个
|
||||
vec.select_nth_unstable_by(max_items, |a, b| {
|
||||
b.1.created_at.cmp(&a.1.created_at)
|
||||
});
|
||||
vec.truncate(max_items);
|
||||
*items = vec.into_iter().collect();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InMemoryStore {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl MemoryStore for InMemoryStore {
|
||||
async fn save(&self, item: MemoryItem) -> Result<(), MemoryError> {
|
||||
{
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.insert(item.id.clone(), item);
|
||||
}
|
||||
self.maybe_evict();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get(&self, id: &str) -> Result<Option<MemoryItem>, MemoryError> {
|
||||
let items = self.items.lock().unwrap();
|
||||
Ok(items.get(id).cloned())
|
||||
}
|
||||
|
||||
async fn delete(&self, id: &str) -> Result<(), MemoryError> {
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.remove(id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list(&self, filter: &MemoryFilter) -> Result<Vec<MemoryItem>, MemoryError> {
|
||||
let items = self.items.lock().unwrap();
|
||||
let mut result: Vec<MemoryItem> = items
|
||||
.values()
|
||||
.filter(|v| match &filter.prefix {
|
||||
Some(p) => v.id.starts_with(p),
|
||||
None => true,
|
||||
})
|
||||
.filter(|v| match filter.since {
|
||||
Some(t) => v.created_at > t,
|
||||
None => true,
|
||||
})
|
||||
.cloned()
|
||||
.collect();
|
||||
// 按 created_at 升序排列（最旧在前）
|
||||
result.sort_by_key(|v| v.created_at);
|
||||
// 应用 offset
|
||||
if let Some(offset) = filter.offset {
|
||||
if offset < result.len() {
|
||||
result.drain(..offset);
|
||||
} else {
|
||||
result.clear();
|
||||
}
|
||||
}
|
||||
// 应用 limit
|
||||
if let Some(limit) = filter.limit {
|
||||
result.truncate(limit);
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
fn make_item(id: &str) -> MemoryItem {
|
||||
MemoryItem {
|
||||
id: id.to_string(),
|
||||
content: format!("content-{id}"),
|
||||
metadata: serde_json::json!({}),
|
||||
created_at: OffsetDateTime::now_utc(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_get_delete_list() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
|
||||
let got = store.get("a").await.unwrap();
|
||||
assert!(got.is_some());
|
||||
assert_eq!(got.unwrap().id, "a");
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
|
||||
store.delete("a").await.unwrap();
|
||||
assert!(store.get("a").await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_is_upsert() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
let mut item = make_item("a");
|
||||
item.content = "updated".to_string();
|
||||
store.save(item).await.unwrap();
|
||||
let got = store.get("a").await.unwrap().unwrap();
|
||||
assert_eq!(got.content, "updated");
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_with_prefix_and_limit() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("foo_a")).await.unwrap();
|
||||
store.save(make_item("foo_b")).await.unwrap();
|
||||
store.save(make_item("bar_a")).await.unwrap();
|
||||
|
||||
let filter = MemoryFilter {
|
||||
prefix: Some("foo_".to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
let list = store.list(&filter).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
|
||||
let filter = MemoryFilter {
|
||||
prefix: Some("foo_".to_string()),
|
||||
limit: Some(1),
|
||||
..Default::default()
|
||||
};
|
||||
let list = store.list(&filter).await.unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn capacity_eviction() {
|
||||
// 强制每次写入都检查
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Capacity { max_items: 2 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
// 第一条和第二条共存
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
// 第三条写入触发淘汰：a 或 b 之一被淘汰
|
||||
store.save(make_item("c")).await.unwrap();
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
// 留下的应该是 b 和 c（最新的两个）
|
||||
let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect();
|
||||
assert!(ids.contains(&"b"));
|
||||
assert!(ids.contains(&"c"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ttl_eviction() {
|
||||
// TTL 设为 0 会立即过期，但我们想保留 "a" 等待 "b" 写入后被淘汰。
|
||||
// 改用小 TTL + 睡眠：先 save a，sleep，save b 时 a 已过期被淘汰。
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Ttl { ttl_secs: 1 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
// 等待超过 1 秒
|
||||
std::thread::sleep(std::time::Duration::from_millis(1100));
|
||||
// 触发淘汰：a 已超过 ttl_secs=1，应被淘汰
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
// 由于 ttl_secs=1，且 b 刚写入，可能刚好处于临界值。
|
||||
// 我们只断言 list 不包含 "a" 即可。
|
||||
let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect();
|
||||
assert!(
|
||||
!ids.contains(&"a"),
|
||||
"expected 'a' to be evicted, but found in {ids:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn none_policy_no_eviction() {
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::None,
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
for i in 0..100 {
|
||||
store.save(make_item(&format!("item_{i}"))).await.unwrap();
|
||||
}
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,266 @@
|
||||
//! 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
use crate::memory::error::MemoryError;
|
||||
use crate::memory::store::{EvictionConfig, EvictionPolicy, MemoryStore};
|
||||
use crate::memory::types::{MemoryFilter, MemoryItem};
|
||||
|
||||
/// 进程内默认实现 —— 基于 HashMap + Mutex，纯内存。
|
||||
pub struct InMemoryStore {
|
||||
items: Mutex<HashMap<String, MemoryItem>>,
|
||||
eviction: EvictionConfig,
|
||||
/// 自上次淘汰检查以来的写入次数。
|
||||
writes_since_check: Mutex<usize>,
|
||||
}
|
||||
|
||||
impl InMemoryStore {
|
||||
/// 创建一个无淘汰策略的 InMemoryStore。
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
items: Mutex::new(HashMap::new()),
|
||||
eviction: EvictionConfig::default(),
|
||||
writes_since_check: Mutex::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
/// 创建一个带淘汰配置的 InMemoryStore。
|
||||
pub fn with_eviction(eviction: EvictionConfig) -> Self {
|
||||
Self {
|
||||
items: Mutex::new(HashMap::new()),
|
||||
eviction,
|
||||
writes_since_check: Mutex::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
fn maybe_evict(&self) {
|
||||
// 不使用 .lock().await 跨点，先取计数判断是否需要淘汰
|
||||
let should_check = {
|
||||
let mut counter = self.writes_since_check.lock().unwrap();
|
||||
*counter += 1;
|
||||
if *counter >= self.eviction.check_interval {
|
||||
*counter = 0;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
}
|
||||
};
|
||||
if !should_check {
|
||||
return;
|
||||
}
|
||||
|
||||
let policy = self.eviction.policy.clone();
|
||||
match policy {
|
||||
EvictionPolicy::None => {}
|
||||
EvictionPolicy::Ttl { ttl_secs } => {
|
||||
let cutoff = OffsetDateTime::now_utc() - time::Duration::seconds(ttl_secs as i64);
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.retain(|_, v| v.created_at > cutoff);
|
||||
}
|
||||
EvictionPolicy::Capacity { max_items } => {
|
||||
let mut items = self.items.lock().unwrap();
|
||||
if items.len() > max_items {
|
||||
let mut vec: Vec<_> = items.drain().collect();
|
||||
// O(n) 部分排序：保留 created_at 最大的 max_items 个
|
||||
vec.select_nth_unstable_by(max_items, |a, b| {
|
||||
b.1.created_at.cmp(&a.1.created_at)
|
||||
});
|
||||
vec.truncate(max_items);
|
||||
*items = vec.into_iter().collect();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for InMemoryStore {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl MemoryStore for InMemoryStore {
|
||||
async fn save(&self, item: MemoryItem) -> Result<(), MemoryError> {
|
||||
{
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.insert(item.id.clone(), item);
|
||||
}
|
||||
self.maybe_evict();
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn get(&self, id: &str) -> Result<Option<MemoryItem>, MemoryError> {
|
||||
let items = self.items.lock().unwrap();
|
||||
Ok(items.get(id).cloned())
|
||||
}
|
||||
|
||||
async fn delete(&self, id: &str) -> Result<(), MemoryError> {
|
||||
let mut items = self.items.lock().unwrap();
|
||||
items.remove(id);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn list(&self, filter: &MemoryFilter) -> Result<Vec<MemoryItem>, MemoryError> {
|
||||
let items = self.items.lock().unwrap();
|
||||
let mut result: Vec<MemoryItem> = items
|
||||
.values()
|
||||
.filter(|v| match &filter.prefix {
|
||||
Some(p) => v.id.starts_with(p),
|
||||
None => true,
|
||||
})
|
||||
.filter(|v| match filter.since {
|
||||
Some(t) => v.created_at > t,
|
||||
None => true,
|
||||
})
|
||||
.cloned()
|
||||
.collect();
|
||||
// 按 created_at 升序排列（最旧在前）
|
||||
result.sort_by_key(|v| v.created_at);
|
||||
// 应用 offset
|
||||
if let Some(offset) = filter.offset {
|
||||
if offset < result.len() {
|
||||
result.drain(..offset);
|
||||
} else {
|
||||
result.clear();
|
||||
}
|
||||
}
|
||||
// 应用 limit
|
||||
if let Some(limit) = filter.limit {
|
||||
result.truncate(limit);
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use time::OffsetDateTime;
|
||||
|
||||
fn make_item(id: &str) -> MemoryItem {
|
||||
MemoryItem {
|
||||
id: id.to_string(),
|
||||
content: format!("content-{id}"),
|
||||
metadata: serde_json::json!({}),
|
||||
created_at: OffsetDateTime::now_utc(),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_get_delete_list() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
|
||||
let got = store.get("a").await.unwrap();
|
||||
assert!(got.is_some());
|
||||
assert_eq!(got.unwrap().id, "a");
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
|
||||
store.delete("a").await.unwrap();
|
||||
assert!(store.get("a").await.unwrap().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_is_upsert() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
let mut item = make_item("a");
|
||||
item.content = "updated".to_string();
|
||||
store.save(item).await.unwrap();
|
||||
let got = store.get("a").await.unwrap().unwrap();
|
||||
assert_eq!(got.content, "updated");
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_with_prefix_and_limit() {
|
||||
let store = InMemoryStore::new();
|
||||
store.save(make_item("foo_a")).await.unwrap();
|
||||
store.save(make_item("foo_b")).await.unwrap();
|
||||
store.save(make_item("bar_a")).await.unwrap();
|
||||
|
||||
let filter = MemoryFilter {
|
||||
prefix: Some("foo_".to_string()),
|
||||
..Default::default()
|
||||
};
|
||||
let list = store.list(&filter).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
|
||||
let filter = MemoryFilter {
|
||||
prefix: Some("foo_".to_string()),
|
||||
limit: Some(1),
|
||||
..Default::default()
|
||||
};
|
||||
let list = store.list(&filter).await.unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn capacity_eviction() {
|
||||
// 强制每次写入都检查
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Capacity { max_items: 2 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
// 第一条和第二条共存
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
// 第三条写入触发淘汰：a 或 b 之一被淘汰
|
||||
store.save(make_item("c")).await.unwrap();
|
||||
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 2);
|
||||
// 留下的应该是 b 和 c（最新的两个）
|
||||
let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect();
|
||||
assert!(ids.contains(&"b"));
|
||||
assert!(ids.contains(&"c"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn ttl_eviction() {
|
||||
// TTL 设为 0 会立即过期，但我们想保留 "a" 等待 "b" 写入后被淘汰。
|
||||
// 改用小 TTL + 睡眠：先 save a，sleep，save b 时 a 已过期被淘汰。
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::Ttl { ttl_secs: 1 },
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
store.save(make_item("a")).await.unwrap();
|
||||
// 等待超过 1 秒
|
||||
std::thread::sleep(std::time::Duration::from_millis(1100));
|
||||
// 触发淘汰：a 已超过 ttl_secs=1，应被淘汰
|
||||
store.save(make_item("b")).await.unwrap();
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
// 由于 ttl_secs=1，且 b 刚写入，可能刚好处于临界值。
|
||||
// 我们只断言 list 不包含 "a" 即可。
|
||||
let ids: Vec<&str> = list.iter().map(|v| v.id.as_str()).collect();
|
||||
assert!(
|
||||
!ids.contains(&"a"),
|
||||
"expected 'a' to be evicted, but found in {ids:?}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn none_policy_no_eviction() {
|
||||
let eviction = EvictionConfig {
|
||||
policy: EvictionPolicy::None,
|
||||
check_interval: 1,
|
||||
};
|
||||
let store = InMemoryStore::with_eviction(eviction);
|
||||
for i in 0..100 {
|
||||
store.save(make_item(&format!("item_{i}"))).await.unwrap();
|
||||
}
|
||||
let list = store.list(&MemoryFilter::default()).await.unwrap();
|
||||
assert_eq!(list.len(), 100);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user