datum-cdc 0.10.1

PostgreSQL logical-replication CDC sources for Datum streams
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

use std::{future::Future, sync::OnceLock};

use tokio::{
    runtime::{Builder, Handle, Runtime},
    task::JoinHandle,
};

static CDC_RUNTIME: OnceLock<Runtime> = OnceLock::new();

pub(crate) fn handle() -> Handle {
    if let Ok(handle) = Handle::try_current() {
        return handle;
    }
    CDC_RUNTIME.get_or_init(build_runtime).handle().clone()
}

pub(crate) fn spawn<F>(future: F) -> JoinHandle<F::Output>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,
{
    handle().spawn(future)
}

fn build_runtime() -> Runtime {
    let worker_threads = std::thread::available_parallelism()
        .map(usize::from)
        .unwrap_or(2)
        .clamp(2, 4);
    Builder::new_multi_thread()
        .worker_threads(worker_threads)
        .thread_name("datum-cdc")
        .enable_io()
        .enable_time()
        .build()
        .expect("failed to build datum-cdc Tokio runtime")
}