1#![cfg_attr(not(debug_assertions), deny(clippy::disallowed_methods))]
4#![cfg_attr(debug_assertions, warn(clippy::disallowed_methods))]
5#![allow(clippy::tabs_in_doc_comments)]
6
7#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
8#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
9#[cfg_attr(any(feature = "http", feature = "ws"), serde(rename_all = "lowercase"))]
10pub enum WireFormat {
11 #[default]
12 Frames,
13 Rbcf,
14}
15
16#[cfg(any(feature = "ws", feature = "grpc"))]
17mod changes;
18#[cfg(any(feature = "ws", feature = "grpc"))]
19pub mod client;
20#[cfg(all(feature = "dst", reifydb_single_threaded))]
21pub mod dst;
22#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
23pub mod error;
24#[cfg(feature = "grpc")]
25pub mod grpc;
26#[cfg(feature = "http")]
27pub mod http;
28#[cfg(any(feature = "ws", feature = "grpc"))]
29mod reconnect;
30#[cfg(any(feature = "http", feature = "ws"))]
31mod session;
32#[cfg(any(feature = "ws", feature = "grpc", all(feature = "dst", reifydb_single_threaded)))]
33pub mod subscription;
34#[cfg(feature = "ws")]
35mod utils;
36#[cfg(feature = "ws")]
37pub mod ws;
38
39#[cfg(any(feature = "http", feature = "ws"))]
40use std::collections::HashMap;
41#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
42use std::sync::Arc;
43
44#[cfg(all(feature = "dst", reifydb_single_threaded))]
45pub use dst::DstClient;
46#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
47pub use error::ClientError;
48#[cfg(feature = "grpc")]
49pub use grpc::{
50 BatchFramesEnvelope, BatchGrpcSubscription, BatchMemberHandle, BatchStreamEvent, GrpcChange, GrpcClient,
51 GrpcClientOptions, GrpcSubscription, RawChangePayload,
52};
53#[cfg(feature = "http")]
54pub use http::HttpClient;
55pub use reifydb_client_derive::FromFrame;
56pub use reifydb_value as value;
57#[cfg(any(feature = "ws", feature = "grpc"))]
58use reifydb_value::error::Error;
59pub use reifydb_value::{
60 params::Params,
61 value::{
62 Value,
63 frame::{
64 column::FrameColumn,
65 data::FrameColumnData,
66 extract::FrameError,
67 frame::Frame,
68 from_frame::FromFrameError,
69 row::{FrameRow, FrameRows},
70 },
71 iso::{IsoDate, IsoDateTime, IsoDuration, IsoTime},
72 ordered_f32::OrderedF32,
73 ordered_f64::OrderedF64,
74 try_from::{FromValueError, TryFromValue, TryFromValueCoerce},
75 value_type::ValueType,
76 },
77};
78#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
79use serde::{Deserialize, Serialize};
80use serde_json::Value as JsonValue;
81#[cfg(any(feature = "ws", feature = "grpc", all(feature = "dst", reifydb_single_threaded)))]
82pub use subscription::{BatchItem, HydrationConfig, Linger, SubscriptionConfig, Throttle, build_subscription_rql};
83#[cfg(feature = "ws")]
84pub use ws::{WsBatchSubscription, WsClient, WsClientOptions};
85
86#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
87#[derive(Debug, Clone)]
88pub struct ResponseMeta {
89 pub fingerprint: String,
90 pub duration: String,
91}
92
93#[derive(Debug)]
94pub struct AdminResult {
95 pub frames: Vec<Frame>,
96 pub meta: Option<ResponseMeta>,
97}
98
99#[derive(Debug)]
100pub struct CommandResult {
101 pub frames: Vec<Frame>,
102 pub meta: Option<ResponseMeta>,
103}
104
105#[derive(Debug)]
106pub struct QueryResult {
107 pub frames: Vec<Frame>,
108 pub meta: Option<ResponseMeta>,
109}
110
111#[derive(Debug, Clone)]
112pub struct LoginResult {
113 pub token: String,
114 pub identity: String,
115}
116
117#[cfg(any(feature = "ws", feature = "grpc"))]
118pub fn connection_lost_error() -> Error {
119 ClientError::ConnectionLost.into()
120}
121
122#[cfg(any(feature = "ws", feature = "grpc"))]
123#[derive(Clone)]
124pub struct ReconnectOptions {
125 pub max_reconnect_attempts: u32,
126 pub reconnect_delay_ms: u64,
127 pub connect_timeout_ms: u64,
128 pub on_disconnect: Option<Arc<dyn Fn() + Send + Sync>>,
129 pub on_reconnect: Option<Arc<dyn Fn() + Send + Sync>>,
130}
131
132#[cfg(any(feature = "ws", feature = "grpc"))]
133impl Default for ReconnectOptions {
134 fn default() -> Self {
135 Self {
136 max_reconnect_attempts: 5,
137 reconnect_delay_ms: 1000,
138 connect_timeout_ms: 30_000,
139 on_disconnect: None,
140 on_reconnect: None,
141 }
142 }
143}
144
145#[cfg(any(feature = "http", feature = "ws"))]
146#[derive(Debug, Serialize, Deserialize)]
147pub struct WireValue {
148 #[serde(rename = "type")]
149 pub type_name: String,
150 pub value: String,
151}
152
153#[cfg(any(feature = "http", feature = "ws"))]
154#[derive(Debug, Serialize, Deserialize)]
155#[serde(untagged)]
156pub enum WireParams {
157 Positional(Vec<WireValue>),
158 Named(HashMap<String, WireValue>),
159}
160
161#[cfg(any(feature = "http", feature = "ws"))]
162fn value_to_wire(value: Value) -> WireValue {
163 let (type_name, value_str): (&str, String) = match &value {
164 Value::None {
165 ..
166 } => ("None", "\u{27EA}none\u{27EB}".to_string()),
167 Value::Boolean(b) => ("Boolean", b.to_string()),
168 Value::Float4(f) => ("Float4", f.to_string()),
169 Value::Float8(f) => ("Float8", f.to_string()),
170 Value::Int1(i) => ("Int1", i.to_string()),
171 Value::Int2(i) => ("Int2", i.to_string()),
172 Value::Int4(i) => ("Int4", i.to_string()),
173 Value::Int8(i) => ("Int8", i.to_string()),
174 Value::Int16(i) => ("Int16", i.to_string()),
175 Value::Utf8(s) => ("Utf8", s.clone()),
176 Value::Uint1(u) => ("Uint1", u.to_string()),
177 Value::Uint2(u) => ("Uint2", u.to_string()),
178 Value::Uint4(u) => ("Uint4", u.to_string()),
179 Value::Uint8(u) => ("Uint8", u.to_string()),
180 Value::Uint16(u) => ("Uint16", u.to_string()),
181 Value::Uuid4(u) => ("Uuid4", u.to_string()),
182 Value::Uuid7(u) => ("Uuid7", u.to_string()),
183 Value::Date(d) => ("Date", d.to_string()),
184 Value::DateTime(dt) => ("DateTime", dt.to_string()),
185 Value::Time(t) => ("Time", t.to_string()),
186 Value::Duration(d) => ("Duration", d.to_iso_string()),
187 Value::Blob(b) => ("Blob", b.to_hex()),
188 Value::IdentityId(id) => ("IdentityId", id.to_string()),
189 Value::Int(i) => ("Int", i.to_string()),
190 Value::Uint(u) => ("Uint", u.to_string()),
191 Value::Decimal(d) => ("Decimal", d.to_string()),
192 Value::Any(v) => return value_to_wire(*v.clone()),
193 Value::DictionaryId(id) => ("DictionaryId", id.to_string()),
194 Value::Type(t) => ("ValueType", t.to_string()),
195 Value::List(items) => ("List", format!("{}", Value::List(items.clone()))),
196 Value::Record(fields) => ("Record", format!("{}", Value::Record(fields.clone()))),
197 Value::Tuple(items) => ("Tuple", format!("{}", Value::Tuple(items.clone()))),
198 };
199 WireValue {
200 type_name: type_name.to_string(),
201 value: value_str,
202 }
203}
204
205#[cfg(any(feature = "http", feature = "ws"))]
206pub fn params_to_wire(params: Params) -> Option<WireParams> {
207 match params {
208 Params::None => None,
209 Params::Positional(values) => Some(WireParams::Positional(
210 Arc::unwrap_or_clone(values).into_iter().map(value_to_wire).collect(),
211 )),
212 Params::Named(map) => Some(WireParams::Named(
213 Arc::unwrap_or_clone(map).into_iter().map(|(k, v)| (k, value_to_wire(v))).collect(),
214 )),
215 }
216}
217
218#[cfg(any(feature = "http", feature = "ws"))]
219#[derive(Debug, Serialize, Deserialize)]
220pub struct Request {
221 pub id: String,
222 #[serde(flatten)]
223 pub payload: RequestPayload,
224}
225
226#[cfg(any(feature = "http", feature = "ws"))]
227#[derive(Debug, Serialize, Deserialize)]
228#[serde(tag = "type", content = "payload")]
229pub enum RequestPayload {
230 Auth(AuthRequest),
231 Admin(AdminRequest),
232 Command(CommandRequest),
233 Query(QueryRequest),
234 Subscribe(SubscribeRequest),
235 Unsubscribe(UnsubscribeRequest),
236 BatchSubscribe(BatchSubscribeRequest),
237 BatchUnsubscribe(BatchUnsubscribeRequest),
238 Call(CallRequest),
239 QueueClaim(WsQueueClaimRequest),
240 Logout,
241}
242
243#[cfg(any(feature = "http", feature = "ws"))]
244#[derive(Debug, Serialize, Deserialize)]
245pub struct AdminRequest {
246 pub rql: String,
247 pub params: Option<WireParams>,
248 #[serde(skip_serializing_if = "Option::is_none")]
249 pub format: Option<WireFormat>,
250}
251
252#[cfg(any(feature = "http", feature = "ws"))]
253#[derive(Debug, Serialize, Deserialize)]
254pub struct AuthRequest {
255 #[serde(skip_serializing_if = "Option::is_none")]
256 pub token: Option<String>,
257 #[serde(skip_serializing_if = "Option::is_none")]
258 pub method: Option<String>,
259 #[serde(skip_serializing_if = "Option::is_none")]
260 pub credentials: Option<HashMap<String, String>>,
261}
262
263#[cfg(any(feature = "http", feature = "ws"))]
264#[derive(Debug, Serialize, Deserialize)]
265pub struct CommandRequest {
266 pub rql: String,
267 pub params: Option<WireParams>,
268 #[serde(skip_serializing_if = "Option::is_none")]
269 pub format: Option<WireFormat>,
270}
271
272#[cfg(any(feature = "http", feature = "ws"))]
273#[derive(Debug, Serialize, Deserialize)]
274pub struct QueryRequest {
275 pub rql: String,
276 pub params: Option<WireParams>,
277 #[serde(skip_serializing_if = "Option::is_none")]
278 pub format: Option<WireFormat>,
279}
280
281#[cfg(any(feature = "http", feature = "ws"))]
282#[derive(Debug, Serialize, Deserialize)]
283pub struct SubscribeRequest {
284 pub rql: String,
285 #[serde(skip_serializing_if = "Option::is_none")]
286 pub format: Option<WireFormat>,
287}
288
289#[cfg(any(feature = "http", feature = "ws"))]
290#[derive(Debug, Serialize, Deserialize)]
291pub struct UnsubscribeRequest {
292 pub subscription_id: String,
293}
294
295#[cfg(any(feature = "http", feature = "ws"))]
296#[derive(Debug, Serialize, Deserialize)]
297pub struct BatchSubscribeRequest {
298 pub queries: Vec<String>,
299 #[serde(skip_serializing_if = "Option::is_none")]
300 pub format: Option<WireFormat>,
301}
302
303#[cfg(any(feature = "http", feature = "ws"))]
304#[derive(Debug, Serialize, Deserialize)]
305pub struct BatchUnsubscribeRequest {
306 pub batch_id: String,
307}
308
309#[cfg(any(feature = "http", feature = "ws"))]
310#[derive(Debug, Serialize, Deserialize)]
311pub struct CallRequest {
312 pub name: String,
313 pub params: Option<WireParams>,
314 #[serde(skip_serializing_if = "Option::is_none")]
315 pub format: Option<WireFormat>,
316}
317
318#[cfg(any(feature = "http", feature = "ws", feature = "grpc"))]
319#[derive(Debug, Clone, Default, Serialize, Deserialize)]
320pub struct QueueClaimRequest {
321 pub queue: String,
322 pub worker: String,
323 #[serde(skip_serializing_if = "Option::is_none")]
324 pub max_n: Option<u32>,
325 #[serde(skip_serializing_if = "Option::is_none")]
326 pub lease_ttl: Option<String>,
327 #[serde(skip_serializing_if = "Option::is_none")]
328 pub wait_for: Option<String>,
329}
330
331#[cfg(any(feature = "http", feature = "ws"))]
332#[derive(Debug, Serialize, Deserialize)]
333pub struct WsQueueClaimRequest {
334 pub queue: String,
335 pub worker: String,
336 #[serde(skip_serializing_if = "Option::is_none")]
337 pub max_n: Option<u32>,
338 #[serde(skip_serializing_if = "Option::is_none")]
339 pub lease_ttl: Option<String>,
340 #[serde(skip_serializing_if = "Option::is_none")]
341 pub wait_for: Option<String>,
342 #[serde(skip_serializing_if = "Option::is_none")]
343 pub format: Option<WireFormat>,
344}
345
346#[cfg(any(feature = "http", feature = "ws"))]
347#[derive(Debug, Serialize, Deserialize)]
348pub struct Response {
349 pub id: String,
350 #[serde(flatten)]
351 pub payload: ResponsePayload,
352}
353
354#[cfg(any(feature = "http", feature = "ws"))]
355#[derive(Debug, Serialize, Deserialize)]
356#[serde(tag = "type", content = "payload")]
357pub enum ResponsePayload {
358 Auth(AuthResponse),
359 Err(ErrResponse),
360 Admin(AdminResponse),
361 Command(CommandResponse),
362 Query(QueryResponse),
363 Subscribed(SubscribedResponse),
364 Unsubscribed(UnsubscribedResponse),
365 BatchSubscribed(BatchSubscribedResponse),
366 BatchUnsubscribed(BatchUnsubscribedResponse),
367 Call(CallResponse),
368 Logout(LogoutResponsePayload),
369}
370
371#[cfg(any(feature = "http", feature = "ws"))]
372#[derive(Debug, Serialize, Deserialize)]
373pub struct AdminResponse {
374 pub content_type: String,
375 pub body: JsonValue,
376 #[serde(default)]
377 pub meta: Option<ResponseMeta>,
378}
379
380#[cfg(any(feature = "http", feature = "ws"))]
381use reifydb_value::error::Diagnostic;
382
383#[cfg(any(feature = "http", feature = "ws"))]
384#[derive(Debug, Serialize, Deserialize)]
385pub struct AuthResponse {
386 #[serde(skip_serializing_if = "Option::is_none")]
387 pub status: Option<String>,
388 #[serde(skip_serializing_if = "Option::is_none")]
389 pub token: Option<String>,
390 #[serde(skip_serializing_if = "Option::is_none")]
391 pub identity: Option<String>,
392}
393
394#[cfg(any(feature = "http", feature = "ws"))]
395#[derive(Debug, Serialize, Deserialize)]
396pub struct ErrResponse {
397 pub diagnostic: Diagnostic,
398}
399
400#[cfg(any(feature = "http", feature = "ws"))]
401#[derive(Debug, Serialize, Deserialize)]
402pub struct CommandResponse {
403 pub content_type: String,
404 pub body: JsonValue,
405 #[serde(default)]
406 pub meta: Option<ResponseMeta>,
407}
408
409#[cfg(any(feature = "http", feature = "ws"))]
410#[derive(Debug, Serialize, Deserialize)]
411pub struct QueryResponse {
412 pub content_type: String,
413 pub body: JsonValue,
414 #[serde(default)]
415 pub meta: Option<ResponseMeta>,
416}
417
418#[cfg(any(feature = "http", feature = "ws"))]
419#[derive(Debug, Serialize, Deserialize)]
420pub struct CallResponse {
421 pub content_type: String,
422 pub body: JsonValue,
423 #[serde(default)]
424 pub meta: Option<ResponseMeta>,
425}
426
427#[cfg(any(feature = "http", feature = "ws"))]
428#[derive(Debug, Serialize, Deserialize)]
429pub struct SubscribedResponse {
430 pub subscription_id: String,
431}
432
433#[cfg(any(feature = "http", feature = "ws"))]
434#[derive(Debug, Serialize, Deserialize)]
435pub struct UnsubscribedResponse {
436 pub subscription_id: String,
437}
438
439#[cfg(any(feature = "http", feature = "ws"))]
440#[derive(Debug, Serialize, Deserialize)]
441pub struct BatchSubscribedResponse {
442 pub batch_id: String,
443 pub members: Vec<BatchMemberInfo>,
444}
445
446#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
447#[derive(Debug, Clone)]
448pub struct BatchMemberInfo {
449 pub index: usize,
450 pub subscription_id: String,
451}
452
453#[cfg(any(feature = "http", feature = "ws"))]
454#[derive(Debug, Serialize, Deserialize)]
455pub struct BatchUnsubscribedResponse {
456 pub batch_id: String,
457}
458
459#[cfg(any(feature = "http", feature = "ws"))]
460#[derive(Debug, Serialize, Deserialize)]
461pub struct LogoutResponsePayload {
462 pub status: String,
463}
464
465#[cfg(any(feature = "http", feature = "ws"))]
466#[derive(Debug, Serialize, Deserialize)]
467#[serde(tag = "type", content = "payload")]
468pub enum ServerPush {
469 Change(WireChangePayload),
470 BatchChange(WireBatchChangePayload),
471 BatchMemberClosed(BatchMemberClosedPayload),
472 BatchClosed(BatchClosedPayload),
473}
474
475#[cfg(any(feature = "http", feature = "ws"))]
476#[derive(Debug, Clone, Serialize, Deserialize)]
477pub struct WireChangePayload {
478 pub subscription_id: String,
479 pub content_type: String,
480 pub body: JsonValue,
481}
482
483#[cfg(any(feature = "http", feature = "ws"))]
484#[derive(Debug, Clone, Serialize, Deserialize)]
485pub struct WireBatchChangePayload {
486 pub batch_id: String,
487 pub entries: Vec<WireBatchChangeEntry>,
488}
489
490#[cfg(any(feature = "http", feature = "ws"))]
491#[derive(Debug, Clone, Serialize, Deserialize)]
492pub struct WireBatchChangeEntry {
493 pub subscription_id: String,
494 pub content_type: String,
495 pub body: JsonValue,
496}
497
498#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
499#[derive(Debug, Clone, Copy, PartialEq, Eq)]
500pub enum ChangeKind {
501 Insert,
502 Update,
503 Remove,
504}
505
506#[derive(Debug, Clone)]
507pub struct FrameChange {
508 pub kind: ChangeKind,
509 pub frame: Frame,
510}
511
512#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
513#[derive(Debug, Clone)]
514pub struct ChangePayload {
515 pub subscription_id: String,
516 pub content_type: String,
517 pub body: JsonValue,
518 #[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
519 pub changes: Vec<FrameChange>,
520}
521
522#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
523#[derive(Debug, Clone)]
524pub struct BatchChangePayload {
525 pub batch_id: String,
526 pub entries: Vec<BatchChangeEntry>,
527}
528
529#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
530#[derive(Debug, Clone)]
531pub struct BatchChangeEntry {
532 pub subscription_id: String,
533 pub content_type: String,
534 pub body: JsonValue,
535 #[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
536 pub changes: Vec<FrameChange>,
537 #[cfg_attr(any(feature = "http", feature = "ws"), serde(skip, default))]
538 pub decode_error: Option<String>,
539}
540
541#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
542#[derive(Debug, Clone)]
543pub struct BatchMemberClosedPayload {
544 pub batch_id: String,
545 pub subscription_id: String,
546}
547
548#[cfg_attr(any(feature = "http", feature = "ws"), derive(Serialize, Deserialize))]
549#[derive(Debug, Clone)]
550pub struct BatchClosedPayload {
551 pub batch_id: String,
552}
553
554#[derive(Debug, Clone)]
555pub enum BatchPushEvent {
556 Change(BatchChangePayload),
557 MemberClosed(BatchMemberClosedPayload),
558 Closed(BatchClosedPayload),
559}