reifydb_remote_proxy/
lib.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4//! Subscribes a local engine to a remote ReifyDB instance, proxying raw RBCF payloads through a caller-supplied
5//! conversion callback.
6//!
7//! The only place external wire-format payloads become local engine events; converting elsewhere would couple
8//! unrelated subsystems to the gRPC client and the wire-format decoders.
9
10#![cfg_attr(not(debug_assertions), deny(clippy::disallowed_methods))]
11#![cfg_attr(debug_assertions, warn(clippy::disallowed_methods))]
12#![cfg_attr(not(debug_assertions), deny(warnings))]
13
14use std::fmt;
15
16use reifydb_client::{GrpcClient, GrpcSubscription, RawChangePayload, SubscriptionConfig, WireFormat};
17use tokio::{
18	select,
19	sync::{mpsc, watch},
20};
21
22#[derive(Debug)]
23pub enum RemoteSubscriptionError {
24	Connect(String),
25	Subscribe(String),
26}
27
28impl fmt::Display for RemoteSubscriptionError {
29	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
30		match self {
31			Self::Connect(e) => write!(f, "Failed to connect to remote: {}", e),
32			Self::Subscribe(e) => write!(f, "Remote subscribe failed: {}", e),
33		}
34	}
35}
36
37pub struct RemoteSubscription {
38	inner: GrpcSubscription,
39	subscription_id: String,
40}
41
42impl RemoteSubscription {
43	pub fn subscription_id(&self) -> &str {
44		&self.subscription_id
45	}
46}
47
48pub async fn connect_remote(
49	address: &str,
50	body: &str,
51	config: SubscriptionConfig,
52	token: Option<&str>,
53	wire_format: WireFormat,
54) -> Result<RemoteSubscription, RemoteSubscriptionError> {
55	let mut client = GrpcClient::connect(address, wire_format)
56		.await
57		.map_err(|e| RemoteSubscriptionError::Connect(e.to_string()))?;
58	if let Some(t) = token {
59		client.authenticate(t);
60	}
61	let sub =
62		client.subscribe(body, config).await.map_err(|e| RemoteSubscriptionError::Subscribe(e.to_string()))?;
63	let subscription_id = sub.subscription_id().to_string();
64	Ok(RemoteSubscription {
65		inner: sub,
66		subscription_id,
67	})
68}
69
70pub async fn proxy_remote<T, F>(
71	mut remote_sub: RemoteSubscription,
72	sender: mpsc::UnboundedSender<T>,
73	mut shutdown: watch::Receiver<bool>,
74	convert: F,
75) where
76	T: Send + 'static,
77	F: Fn(RawChangePayload) -> T,
78{
79	loop {
80		select! {
81			payload = remote_sub.inner.recv_raw() => {
82				match payload {
83					Some(payload) => {
84						if sender.send(convert(payload)).is_err() {
85							break;
86						}
87					}
88					None => break,
89				}
90			}
91			_ = sender.closed() => break,
92			_ = shutdown.changed() => break,
93		}
94	}
95}
96
97pub async fn proxy_remote_to_sink<F>(
98	mut remote_sub: RemoteSubscription,
99	mut shutdown: watch::Receiver<bool>,
100	mut sink: F,
101) where
102	F: FnMut(RawChangePayload) -> bool + Send + 'static,
103{
104	loop {
105		select! {
106			payload = remote_sub.inner.recv_raw() => {
107				match payload {
108					Some(payload) => {
109						if !sink(payload) {
110							break;
111						}
112					}
113					None => break,
114				}
115			}
116			_ = shutdown.changed() => break,
117		}
118	}
119}
reifydb_remote_proxy/lib.rs

reifydb_remote_proxy/
lib.rs
