Skip to content

Commit fe120f7

Browse files
smiggleworthsmiggleworth
committed
wip
1 parent ffc7585 commit fe120f7

11 files changed

Lines changed: 126 additions & 69 deletions

File tree

‎src/core/control/handler.rs‎

Lines changed: 6 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,7 @@
11
// Control domain handler - routes all control:// operations
22

3-
use crate::core::domain::{Domain, DomainRequest, DomainResponse, SubSender};
3+
use crate::core::domain::{Domain, DomainRequest, DomainResponse};
44
use crate::protocol::tags::{TAG_BODY, TAG_ERR_MSG, TAG_ID, TAG_ROUTE};
5-
use crate::storage::traits::KvStore;
65
use super::service::ControlService;
76
use super::types::ControlOperation;
87
use crate::core::notice::NoticeService;
@@ -113,7 +112,6 @@ impl Domain for ControlDomain {
113112
fn handle<'a>(
114113
&'a self,
115114
request: DomainRequest,
116-
_kv_store: Arc<dyn KvStore>,
117115
) -> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
118116
Box::pin(async move {
119117
// Parse body from TLV payload
@@ -248,7 +246,6 @@ mod tests {
248246
async fn should_handle_heartbeat_operation() {
249247
// Arrange
250248
let domain = ControlDomain::new();
251-
let store = Arc::new(crate::storage::mem::MemStore::new()) as Arc<dyn KvStore>;
252249
let mut payload = Vec::new();
253250
payload.push(TAG_BODY);
254251
let body = b"{\"nodeId\":\"test-node\",\"timestamp\":1234567890}";
@@ -270,7 +267,7 @@ mod tests {
270267
};
271268

272269
// Act
273-
let response = domain.handle(request, store).await;
270+
let response = domain.handle(request).await;
274271

275272
// Assert
276273
match response {
@@ -285,7 +282,6 @@ mod tests {
285282
async fn should_handle_shutdown_operation() {
286283
// Arrange
287284
let domain = ControlDomain::new();
288-
let store = Arc::new(crate::storage::mem::MemStore::new()) as Arc<dyn KvStore>;
289285
let mut payload = Vec::new();
290286
payload.push(TAG_BODY);
291287
let body = b"{\"nodeId\":\"test-node\",\"reason\":\"maintenance\"}";
@@ -307,7 +303,7 @@ mod tests {
307303
};
308304

309305
// Act
310-
let response = domain.handle(request, store).await;
306+
let response = domain.handle(request).await;
311307

312308
// Assert
313309
match response {
@@ -322,7 +318,6 @@ mod tests {
322318
async fn should_return_error_when_body_missing() {
323319
// Arrange
324320
let domain = ControlDomain::new();
325-
let store = Arc::new(crate::storage::mem::MemStore::new()) as Arc<dyn KvStore>;
326321
let payload = Vec::new(); // Empty payload
327322

328323
let request = DomainRequest {
@@ -340,11 +335,12 @@ mod tests {
340335
};
341336

342337
// Act
343-
let response = domain.handle(request, store).await;
338+
let response = domain.handle(request).await;
344339

345340
// Assert
346341
match response {
347-
DomainResponse::Frame(frame) => {
342+
DomainResponse::Frame(_frame) => {
343+
// Success - error frame returned
348344
}
349345
_ => panic!("Expected Frame response with error"),
350346
}

‎src/core/domain.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,6 @@
22
// to handle all operations for its scheme
33

44
use crate::protocol::route::Route;
5-
use crate::storage::traits::KvStore;
6-
use std::sync::Arc;
75
use tokio::sync::mpsc;
86

97
/// Type alias for subscriber channels (used by domains that support pub/sub)
@@ -42,7 +40,9 @@ pub trait Domain: Send + Sync {
4240
/// Handle a request for this domain
4341
/// Domain parses TLV tags from request.payload to extract operation details
4442
/// Returns DomainResponse with TLV-encoded response or error
45-
fn handle<'a>(&'a self, request: DomainRequest, kv_store: Arc<dyn KvStore>)
43+
///
44+
/// Domains that need persistent storage should manage their own KvStore instance
45+
fn handle<'a>(&'a self, request: DomainRequest)
4646
-> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>>;
4747

4848
/// Get the scheme(s) this domain handles (e.g., "queue", "kv", "stream")

‎src/core/engine.rs‎

Lines changed: 30 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@ use tokio::task::JoinHandle;
77
use crate::core::domain::{Domain, DomainRequest, DomainResponse};
88
use crate::core::router::Router;
99
use crate::protocol::route::parse_route;
10-
use crate::storage::traits::KvStore;
1110

1211
// Keep the subscription sender type for compatibility
1312
pub type SubSender = mpsc::Sender<(
@@ -293,18 +292,39 @@ impl EngineHandle {
293292
}
294293

295294
/// Start the engine task with domain handlers
296-
pub fn start_engine(kv_store: Arc<dyn KvStore>) -> EngineHandle {
297-
let (handle, _jh) = start_engine_with_join(kv_store);
295+
pub fn start_engine() -> EngineHandle {
296+
let (_jh, handle) = start_engine_with_join();
298297
handle
299298
}
300299

301-
pub fn start_engine_with_join(kv_store: Arc<dyn KvStore>) -> (EngineHandle, JoinHandle<()>) {
300+
pub fn start_engine_with_join() -> (JoinHandle<()>, EngineHandle) {
302301
let (tx, mut rx) = mpsc::channel::<EngineCommand>(1024);
303302
let handle = EngineHandle::new(tx.clone());
304303

305304
// Create domain handlers as Arc for shared ownership
306305
let mut domains: HashMap<&'static str, Arc<dyn Domain>> = HashMap::new();
307306

307+
// Create a mock KV store for domains that need storage
308+
// TODO: Replace with proper storage backend
309+
use crate::storage::traits::{KvStore, KvTransaction};
310+
use bytes::Bytes;
311+
312+
#[derive(Clone)]
313+
struct MockStore;
314+
impl KvStore for MockStore {
315+
fn put(&self, _key: &[u8], _value: &[u8]) -> Result<(), String> { Ok(()) }
316+
fn get(&self, _key: &[u8]) -> Result<Option<Bytes>, String> { Ok(None) }
317+
fn delete(&self, _key: &[u8]) -> Result<(), String> { Ok(()) }
318+
fn put_batch(&self, _writes: Vec<(Vec<u8>, Vec<u8>)>) -> Result<(), String> { Ok(()) }
319+
fn delete_batch(&self, _keys: Vec<Vec<u8>>) -> Result<(), String> { Ok(()) }
320+
fn scan(&self, _start: &[u8], _end: &[u8]) -> Result<Vec<(Bytes, Bytes)>, String> { Ok(vec![]) }
321+
fn flush(&self) -> Result<(), String> { Ok(()) }
322+
fn begin_transaction(&self) -> Result<Box<dyn KvTransaction>, String> {
323+
Err("Transactions not supported in mock".to_string())
324+
}
325+
}
326+
let kv_store = Arc::new(MockStore) as Arc<dyn KvStore>;
327+
308328
// Register all domains
309329
use crate::core::{control::ControlDomain, kv::KvDomain, lease::LeaseDomain,
310330
notice::NoticeDomain, queue::QueueDomain, rpc::RpcDomain,
@@ -313,11 +333,11 @@ pub fn start_engine_with_join(kv_store: Arc<dyn KvStore>) -> (EngineHandle, Join
313333
// Queue domain
314334
domains.insert("queue", Arc::new(QueueDomain::new()));
315335

316-
// KV domain
317-
domains.insert("kv", Arc::new(KvDomain::new()));
336+
// KV domain - needs storage
337+
domains.insert("kv", Arc::new(KvDomain::new(Arc::clone(&kv_store))));
318338

319-
// Stream domain
320-
domains.insert("stream", Arc::new(StreamDomain::new()));
339+
// Stream domain - needs storage
340+
domains.insert("stream", Arc::new(StreamDomain::new(Arc::clone(&kv_store))));
321341

322342
// Lease domain
323343
domains.insert("lease", Arc::new(LeaseDomain::new()));
@@ -374,7 +394,7 @@ pub fn start_engine_with_join(kv_store: Arc<dyn KvStore>) -> (EngineHandle, Join
374394
};
375395

376396
// Dispatch to domain
377-
let response = domain.handle(request, kv_store.clone()).await;
397+
let response = domain.handle(request).await;
378398

379399
// Convert domain response to bytes and send response
380400
match response {
@@ -453,5 +473,5 @@ pub fn start_engine_with_join(kv_store: Arc<dyn KvStore>) -> (EngineHandle, Join
453473
}
454474
});
455475

456-
(handle, jh)
476+
(jh, handle)
457477
}

‎src/core/kv/handler.rs‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,13 @@ use crate::protocol::tags::{TAG_BODY, TAG_ERR_MSG, TAG_ID};
66
use crate::storage::traits::KvStore;
77
use std::sync::Arc;
88

9-
pub struct KvDomain;
9+
pub struct KvDomain {
10+
kv_store: Arc<dyn KvStore>,
11+
}
1012

1113
impl KvDomain {
12-
pub fn new() -> Self {
13-
Self
14+
pub fn new(kv_store: Arc<dyn KvStore>) -> Self {
15+
Self { kv_store }
1416
}
1517

1618
/// Parse TLV body to extract key (TAG_ID) and value (TAG_BODY)
@@ -114,15 +116,32 @@ impl KvDomain {
114116

115117
impl Default for KvDomain {
116118
fn default() -> Self {
117-
Self::new()
119+
// For tests - use a mock store
120+
use crate::storage::traits::KvTransaction;
121+
use bytes::Bytes;
122+
123+
struct MockStore;
124+
impl KvStore for MockStore {
125+
fn put(&self, _key: &[u8], _value: &[u8]) -> Result<(), String> { Ok(()) }
126+
fn get(&self, _key: &[u8]) -> Result<Option<Bytes>, String> { Ok(None) }
127+
fn delete(&self, _key: &[u8]) -> Result<(), String> { Ok(()) }
128+
fn put_batch(&self, _writes: Vec<(Vec<u8>, Vec<u8>)>) -> Result<(), String> { Ok(()) }
129+
fn delete_batch(&self, _keys: Vec<Vec<u8>>) -> Result<(), String> { Ok(()) }
130+
fn scan(&self, _start: &[u8], _end: &[u8]) -> Result<Vec<(Bytes, Bytes)>, String> { Ok(vec![]) }
131+
fn flush(&self) -> Result<(), String> { Ok(()) }
132+
fn begin_transaction(&self) -> Result<Box<dyn KvTransaction>, String> {
133+
Err("Transactions not supported in mock".to_string())
134+
}
135+
}
136+
137+
Self::new(Arc::new(MockStore))
118138
}
119139
}
120140

121141
impl Domain for KvDomain {
122142
fn handle<'a>(
123143
&'a self,
124144
request: DomainRequest,
125-
kv_store: Arc<dyn KvStore>,
126145
) -> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
127146
Box::pin(async move {
128147
// Determine operation from route
@@ -139,6 +158,9 @@ impl Domain for KvDomain {
139158
// Use route_str for namespacing
140159
let route_str = &request.route_str;
141160

161+
// Use self.kv_store (clone the Arc for service)
162+
let kv_store = Arc::clone(&self.kv_store);
163+
142164
// Create service with the KV store
143165
let service = super::service::KvService::new(kv_store);
144166

‎src/core/lease/handler.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
// Lease domain handler - routes all lease:// operations
22

33
use crate::core::domain::{Domain, DomainRequest, DomainResponse};
4-
use crate::storage::traits::KvStore;
5-
use std::sync::Arc;
64

75
pub struct LeaseDomain;
86

@@ -19,7 +17,7 @@ impl Default for LeaseDomain {
1917
}
2018

2119
impl Domain for LeaseDomain {
22-
fn handle<'a>(&'a self, _request: DomainRequest, _kv_store: Arc<dyn KvStore>)
20+
fn handle<'a>(&'a self, _request: DomainRequest)
2321
-> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
2422
Box::pin(async move {
2523
panic!("LeaseDomain::handle not yet implemented")

‎src/core/notice/handler.rs‎

Lines changed: 2 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
use crate::core::domain::{Domain, DomainRequest, DomainResponse, SubSender};
44
use crate::protocol::tags::{TAG_BODY, TAG_ERR_MSG, TAG_ID, TAG_ROUTE, TAG_ROUTE_REPLY, TAG_SEQ, TAG_STREAM_END, TAG_SUBSCRIBE, TAG_UNSUBSCRIBE};
5-
use crate::storage::traits::KvStore;
65
use super::service::NoticeService;
76
use std::sync::Arc;
87
use tokio::sync::Mutex;
@@ -142,7 +141,6 @@ impl Domain for NoticeDomain {
142141
fn handle<'a>(
143142
&'a self,
144143
request: DomainRequest,
145-
_kv_store: Arc<dyn KvStore>,
146144
) -> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
147145
Box::pin(async move {
148146
// Determine operation from TLV tags
@@ -306,7 +304,6 @@ mod tests {
306304
async fn should_handle_subscribe_request() {
307305
// Arrange
308306
let domain = NoticeDomain::new();
309-
let store = Arc::new(Mutex::new(MemStore::new()));
310307
let mut payload = Vec::new();
311308
payload.push(TAG_SUBSCRIBE);
312309
payload.push(0);
@@ -326,7 +323,7 @@ mod tests {
326323
};
327324

328325
// Act
329-
let response = domain.handle(request, store).await;
326+
let response = domain.handle(request).await;
330327

331328
// Assert
332329
match response {
@@ -341,7 +338,6 @@ mod tests {
341338
async fn should_handle_publish_request() {
342339
// Arrange
343340
let domain = NoticeDomain::new();
344-
let store = Arc::new(Mutex::new(MemStore::new()));
345341
let mut payload = Vec::new();
346342
payload.push(TAG_ID);
347343
let id = b"msg-1";
@@ -367,7 +363,7 @@ mod tests {
367363
};
368364

369365
// Act
370-
let response = domain.handle(request, store).await;
366+
let response = domain.handle(request).await;
371367

372368
// Assert
373369
match response {

‎src/core/queue/handler.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
// Queue domain handler - routes all queue:// operations
22

33
use crate::core::domain::{Domain, DomainRequest, DomainResponse};
4-
use crate::storage::traits::KvStore;
5-
use std::sync::Arc;
64

75
pub struct QueueDomain;
86

@@ -19,7 +17,7 @@ impl Default for QueueDomain {
1917
}
2018

2119
impl Domain for QueueDomain {
22-
fn handle<'a>(&'a self, _request: DomainRequest, _kv_store: Arc<dyn KvStore>)
20+
fn handle<'a>(&'a self, _request: DomainRequest)
2321
-> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
2422
Box::pin(async move {
2523
// TODO: Implement queue domain logic

‎src/core/rpc/handler.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
// RPC domain handler - routes all rpc:// operations
22

33
use crate::core::domain::{Domain, DomainRequest, DomainResponse};
4-
use crate::storage::traits::KvStore;
5-
use std::sync::Arc;
64

75
pub struct RpcDomain;
86

@@ -19,7 +17,7 @@ impl Default for RpcDomain {
1917
}
2018

2119
impl Domain for RpcDomain {
22-
fn handle<'a>(&'a self, _request: DomainRequest, _kv_store: Arc<dyn KvStore>)
20+
fn handle<'a>(&'a self, _request: DomainRequest)
2321
-> std::pin::Pin<Box<dyn std::future::Future<Output = DomainResponse> + Send + 'a>> {
2422
Box::pin(async move {
2523
panic!("RpcDomain::handle not yet implemented")

0 commit comments

Comments
 (0)