reifydb_cdc/
testing.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::sync::Arc;
5
6use reifydb_catalog::{cache::CatalogCache, catalog::Catalog};
7use reifydb_codec::{key::encoded::EncodedKey, row::bytes::EncodedBytes};
8use reifydb_core::{common::CommitVersion, event::EventBus};
9use reifydb_runtime::{
10	actor::system::ActorSystem,
11	context::{
12		clock::{Clock, MockClock},
13		rng::Rng,
14	},
15	version_epoch::VersionEpoch,
16};
17use reifydb_store_multi::MultiStore;
18use reifydb_store_single::SingleStore;
19use reifydb_transaction::{
20	interceptor::interceptors::Interceptors,
21	multi::transaction::MultiTransaction,
22	single::SingleTransaction,
23	transaction::{command::CommandTransaction, query::QueryTransaction},
24};
25use reifydb_value::{
26	Result,
27	util::cowvec::CowVec,
28	value::{duration::Duration, identity::IdentityId},
29};
30
31use crate::consume::host::CdcHost;
32
33#[derive(Clone)]
34pub struct TestCdcHost {
35	multi: MultiTransaction,
36	single: SingleTransaction,
37	pub event_bus: EventBus,
38	pub catalog: Catalog,
39	pub clock: Clock,
40	pub mock: MockClock,
41}
42
43impl TestCdcHost {
44	pub fn with_clock(initial_nanos: u64) -> Self {
45		let multi_store = MultiStore::testing_memory();
46		let single_store = SingleStore::testing_memory();
47		let actor_system = ActorSystem::testing(Clock::Real);
48		let spawner = actor_system.spawner();
49		let event_bus = EventBus::new(&spawner);
50		let single = SingleTransaction::new(single_store, event_bus.clone());
51		let catalog_cache = CatalogCache::new();
52		let mock = MockClock::new(initial_nanos);
53		let clock = Clock::Mock(mock.clone());
54		let multi = MultiTransaction::new(
55			multi_store,
56			single.clone(),
57			event_bus.clone(),
58			spawner,
59			clock.clone(),
60			VersionEpoch::new(),
61			Rng::seeded(42),
62			Arc::new(catalog_cache.clone()),
63		)
64		.unwrap();
65		Self {
66			multi,
67			single,
68			event_bus,
69			catalog: Catalog::new(catalog_cache),
70			clock,
71			mock,
72		}
73	}
74
75	pub fn new() -> Self {
76		Self::with_clock(1_000_000_000)
77	}
78}
79
80impl Default for TestCdcHost {
81	fn default() -> Self {
82		Self::new()
83	}
84}
85
86impl CdcHost for TestCdcHost {
87	fn begin_command(&self) -> Result<CommandTransaction> {
88		CommandTransaction::new(
89			self.multi.clone(),
90			self.single.clone(),
91			self.event_bus.clone(),
92			Interceptors::new(),
93			IdentityId::system(),
94			self.clock.clone(),
95		)
96	}
97
98	fn begin_query(&self) -> Result<QueryTransaction> {
99		Ok(QueryTransaction::new(self.multi.begin_query()?, self.single.clone(), IdentityId::system()))
100	}
101
102	fn current_version(&self) -> Result<CommitVersion> {
103		Ok(CommitVersion(1))
104	}
105
106	fn done_until(&self) -> CommitVersion {
107		CommitVersion(1)
108	}
109
110	fn cdc_producer_watermark(&self) -> CommitVersion {
111		CommitVersion(1)
112	}
113
114	fn wait_for_mark_timeout(&self, _version: CommitVersion, _timeout: Duration) -> bool {
115		true
116	}
117
118	fn notify_on_mark(&self, _version: CommitVersion, callback: Box<dyn FnOnce() + Send>) {
119		callback();
120	}
121
122	fn catalog(&self) -> &Catalog {
123		&self.catalog
124	}
125}
126
127pub fn make_key(s: &str) -> EncodedKey {
128	EncodedKey::new(s.as_bytes())
129}
130
131pub fn make_bytes(s: &str) -> EncodedBytes {
132	EncodedBytes(CowVec::new(s.as_bytes().to_vec()))
133}
reifydb_cdc/testing.rs

reifydb_cdc/
testing.rs
