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

reifydb_cdc/
testing.rs
