river-data-core

Feed your instrument or spreadsheet data into river-data by implementing two functions.

river-data-core is the shared library for the river-data platform: the sync client and the wire protocol both ends share. A sync service built on it registers its data streams once, then pushes new readings on a schedule. Enrollment, heartbeats, retries, token rotation, batching and remote commands (pause, resume, trigger a full sync from the dashboard) are all handled for you.

Install

[dependencies]
river-data-core = { version = "0.11", features = ["client"] }

Common companion crates (chrono, uuid, serde_json, tracing, async_trait) are re-exported at the crate root, so one dependency is enough to start.

Quick start

A sync service answers two questions about your data source: what streams exist, and what new readings arrived since the last sync. You answer them by implementing the SourceBackend trait. The full program below pushes a synthetic temperature signal; it is examples/minimal_backend.rs in this repository.

First, describe your streams. A stream is one series of readings, ie. one column of a spreadsheet or one sensor channel:

async fn discover_streams(&self) -> Result<Vec<StreamDescriptor>, BackendError> {
    Ok(vec![StreamDescriptor {
        source_key: "demo-temperature".to_string(),
        source_name: "Demo Temperature".to_string(),
        source_path: "demo/lab/temperature".to_string(),
        metadata: json!({ "units": "degC" }),
        measurement_type: Some("continuous".to_string()),
        sensor_id: None,
        replicates: None,
        decimal_places: None,
    }])
}

source_key is the stable identifier on your side (a column name, a location id). Keep it stable across restarts: a changed key registers as a new, empty stream. measurement_type is "continuous" for logger data or "spot" for grab samples. sensor_id names the instrument behind the stream when your backend already knows it; leave it None and the server resolves the instrument from the metadata serial when the stream is imported or paired. replicates declares a replicate family (grab samples taken in several vials at one instant); None for a plain series.

Second, fetch new readings. Each request carries since, the time of the newest reading river-data already has for that stream, so you only return what is new. This is the examples/csv_folder.rs version, one stream per time,value file:

async fn fetch_readings(
    &self,
    requests: &[StreamFetchRequest],
) -> Result<Vec<StreamReadings>, BackendError> {
    let mut out = Vec::new();
    for req in requests {
        let path = self.dir.join(format!("{}.csv", req.source_key));
        let content = std::fs::read_to_string(&path)?;

        // Keep only rows newer than the cursor so re-syncs don't re-send
        let readings: Vec<IngestReading> = content
            .lines()
            .filter_map(parse_row)
            .filter(|(time, _)| req.since.is_none_or(|s| *time > s))
            .map(|(time, value)| IngestReading::new(time, value))
            .collect();

        out.push(StreamReadings::new(
            req.stream_id,
            req.source_key.clone(),
            readings,
        ));
    }
    Ok(out)
}

Third, run it:

fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    run_sync_service(|_config| async { Ok(Box::new(DemoBackend) as Box<dyn SourceBackend>) })
}

Ask a river-data administrator to create sync credentials for you (Sync Services page in the dashboard), then set three environment variables and start the service:

export API_BASE_URL=https://river-data.example.org
export SERVICE_CLIENT_ID=svc_demo
export SERVICE_CLIENT_SECRET=your-secret
cargo run --example minimal_backend --features client

The service enrolls itself on startup, syncs every five minutes, and shows up on the dashboard with its status and per-cycle event log.

One step remains before the data appears anywhere: an administrator pairs each new stream to a site parameter (Streams page in the dashboard). Until then readings are stored but belong to no site, so charts and exports show nothing. Ask for pairing once your streams are registered; syncing continues normally either way and the paired history is backfilled.

Usage

Cursors and full syncs. On the first sync since is None: return your full history (or a sensible window). After that, since is the newest ingested reading per stream. A full sync (triggered from the dashboard) passes None again; re-sent readings are deduplicated server-side, so returning overlap is safe.

Reading files. examples/csv_folder.rs syncs a folder of time,value CSV files, one stream per file (useful as a template for lab exports). It reads the folder named by DATA_DIR (default ./data).

Mutable sources. A source that is edited in place (a portal database) cannot be synced by cursor. Declare reconciled() and attach a completeness window to each fetch; the server diffs stored content against the payload, corrects changed values and withdraws absent rows. examples/reconciled_backend.rs is the template.

Status events. Override fetch_status_events to report device telemetry (battery, signal, reachability) alongside readings. The default reports nothing.

Custom commands. Override handle_command to react to commands sent from the dashboard beyond the built-in sync/pause/resume set.

Real services. river-data-sync-vaisala is an HTTP source with status events; river-data-rshiny reads three MySQL portal schemas behind one binary.

Onboarding a new source. docs/sync-service-onboarding.md covers credentials, the control plane, stream identity, replicate families, standard curves, reconciled sources, pairing and migration, with the two real services as worked examples.

Full control. If the driver's cycle does not fit your source, implement the SyncService trait instead and drive SyncServiceRunner from your own main.

Configuration

Variables are also read from a .env file in the working directory.

Features

• client: sync service runner, driver and HTTP client (reqwest)
• openapi: ToSchema derives on the protocol types, for a host API that publishes a spec

The control plane's server side lives in river-data-api, which owns the database schema those handlers query. This crate defines the protocol they speak.

License

MIT