nodedb 0.0.0-beta.1

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

//! Shared test helpers for CoreLoop integration tests.

use std::time::{Duration, Instant};

use nodedb::bridge::dispatch::{BridgeRequest, BridgeResponse};
use nodedb::bridge::envelope::{PhysicalPlan, Priority, Request, Status};
use nodedb::data::executor::core_loop::CoreLoop;
use nodedb::types::*;
use nodedb_bridge::buffer::{Consumer, Producer, RingBuffer};

pub fn make_core() -> (CoreLoop, Producer<BridgeRequest>, Consumer<BridgeResponse>) {
    let dir = tempfile::tempdir().unwrap();
    let (req_tx, req_rx) = RingBuffer::channel::<BridgeRequest>(64);
    let (resp_tx, resp_rx) = RingBuffer::channel::<BridgeResponse>(64);
    let core = CoreLoop::open(0, req_rx, resp_tx, dir.path()).unwrap();
    std::mem::forget(dir);
    (core, req_tx, resp_rx)
}

pub fn make_request(plan: PhysicalPlan) -> Request {
    Request {
        request_id: RequestId::new(1),
        tenant_id: TenantId::new(1),
        vshard_id: VShardId::new(0),
        plan,
        deadline: Instant::now() + Duration::from_secs(5),
        priority: Priority::Normal,
        trace_id: 0,
        consistency: ReadConsistency::Strong,
        idempotency_key: None,
    }
}

pub fn make_request_with_id(id: u64, plan: PhysicalPlan) -> Request {
    Request {
        request_id: RequestId::new(id),
        ..make_request(plan)
    }
}

/// Push a request, tick, pop response — asserts Ok status.
pub fn send_ok(
    core: &mut CoreLoop,
    req_tx: &mut Producer<BridgeRequest>,
    resp_rx: &mut Consumer<BridgeResponse>,
    plan: PhysicalPlan,
) -> Vec<u8> {
    req_tx
        .try_push(BridgeRequest {
            inner: make_request(plan),
        })
        .unwrap();
    core.tick();
    let resp = resp_rx.try_pop().unwrap();
    assert_eq!(
        resp.inner.status,
        Status::Ok,
        "expected Ok, got {:?}",
        resp.inner.error_code
    );
    resp.inner.payload.to_vec()
}

/// Push a request, tick, pop response — returns the raw response.
pub fn send_raw(
    core: &mut CoreLoop,
    req_tx: &mut Producer<BridgeRequest>,
    resp_rx: &mut Consumer<BridgeResponse>,
    plan: PhysicalPlan,
) -> nodedb::bridge::envelope::Response {
    req_tx
        .try_push(BridgeRequest {
            inner: make_request(plan),
        })
        .unwrap();
    core.tick();
    resp_rx.try_pop().unwrap().inner
}

pub fn payload_json(payload: &[u8]) -> String {
    nodedb::data::executor::response_codec::decode_payload_to_json(payload)
}

pub fn payload_value(payload: &[u8]) -> serde_json::Value {
    let json = payload_json(payload);
    serde_json::from_str(&json).unwrap_or(serde_json::Value::Null)
}