A cross-platform Rust library for real-time property synchronization between devices and services.
Hive lets you define observable properties on one device and automatically keep them in sync across every connected peer, whether they're linked over TCP sockets, Bluetooth, MQTT, or WebSockets. When a property changes on any node, the update propagates to all others.
Properties are the unit of state in Hive. Each property has a name, a typed value (string, integer, float, bool), and a reactive change stream. You can observe changes with callbacks or async streams:
let mut hive = Hive::new_from_str_unknown(r#"
name = "Thermostat"
listen = "3000"
[Properties]
temperature = 72
target_temp = 68.5
fan_active = false
"#);
// React to changes
hive.get_mut_property_by_name("temperature")
.unwrap()
.on_next(|value| {
println!("Temperature changed: {:?}", value);
});
// Or use async streams
let mut stream = hive.get_mut_property_by_name("temperature")
.unwrap()
.stream
.clone();
tokio::spawn(async move {
while let Some(value) = stream.next().await {
println!("Stream: {:?}", value);
}
});Peers connect Hive instances together. A server listens for connections; clients connect to servers. Once linked, all properties synchronize automatically:
// Server
let server = Hive::new_from_str_unknown(r#"
name = "Server"
listen = "3000"
[Properties]
lightValue = 0
"#);
// Client, connects and receives all server properties
let client = Hive::new_from_str_unknown(r#"
name = "Client"
connect = "127.0.0.1:3000"
"#);Handlers let you modify properties and send messages from outside the Hive event loop:
let cancellation_token = CancellationToken::new();
let mut handler = hive.go(true, cancellation_token).await;
// Set a property, propagates to all peers
handler.set_property("lightValue", Some(&42.into())).await;
// Send a direct message to a specific peer
handler.send_to_peer("Client", "hello there").await;| Protocol | Feature Flag | Use Case |
|---|---|---|
| TCP | (always available) | LAN communication between services |
| WebSocket | websock |
Browser and web client integration |
| Bluetooth | bluetooth |
Direct device-to-device without network infrastructure |
| MQTT | mqtt |
IoT broker-based pub/sub across networks |
The MQTT transport connects Hive to any standard MQTT broker (Mosquitto, EMQX, HiveMQ, etc.). Properties map to MQTT topics under a configurable prefix, using a compact binary wire format for payloads:
name = "SensorNode"
[MQTT]
host = "192.168.1.100"
port = 1883
client_id = "hive-sensor-01"
topic_prefix = "hive/home"
qos = 1
keep_alive = 30
subscribe = ["temperature", "humidity"]
[Properties]
temperature = 0
humidity = 0With this config, property changes publish to hive/home/temperature, hive/home/humidity, etc. Any other Hive instance (or MQTT client speaking the binary format) subscribed to the same topics will receive updates.
Hive instances are configured via TOML, either inline strings or files:
// From a string
let hive = Hive::new_from_str_unknown(r#"
name = "MyHive"
listen = "3000"
[Properties]
sensor = 0
"#);
// From a file
let hive = Hive::new("config.toml");# Identity
name = "MyHive"
# TCP, pick one: listen (server) or connect (client)
listen = "3000" # or "192.168.1.10:3000"
connect = "192.168.1.10:3000" # server address
# Optional reliability settings, see "Connection Reliability" below
wave_timeout_ms = 800 # actively ping if a peer goes idle this long
max_message_size_bytes = 1048576 # cap on a single message's claimed size, this is the default
# Bluetooth (requires "bluetooth" feature)
bt_listen = "Hive_Peripheral" # advertise as BLE peripheral
bt_connect = "Hive_Central" # connect as BLE central
# MQTT (requires "mqtt" feature)
[MQTT]
host = "127.0.0.1"
port = 1883
client_id = "hive-node"
topic_prefix = "hive"
qos = 1 # 0 = AtMostOnce, 1 = AtLeastOnce, 2 = ExactlyOnce
keep_alive = 30
subscribe = ["prop1", "prop2"]
# Properties with optional arguments
[Properties]
simple_int = 42
simple_str = "hello"
simple_bool = true
simple_float = 3.14
with_args = { val = 0, rest_set = "http://localhost:8000/save/{val}" }Debouncing (receive side) For high-frequency updates like motor control, properties support a backoff window on their local on_next callback. The first change fires immediately, subsequent changes within the window are absorbed, and the latest value is delivered when the timer expires:
let prop = hive.get_mut_property_by_name("motor_speed").unwrap();
prop.set_backoff(&50); // 50ms debounce window
prop.on_next(|v| {
// Fires at most once per 50ms
send_to_motor(v);
});This only rate-limits the local callback. It does not reduce outgoing network traffic, Handler::set_property builds its own throwaway Property for the wire and never looks at backoff.
Debouncing (send side) To rate-limit how often you actually send updates, e.g. a UI slider firing on every pixel of drag, use Handler::set_property_debounced instead. Same leading+trailing behavior as backoff, applied to the network send itself, keyed per property name:
// First call in a burst sends immediately, further calls within 50ms
// coalesce down to the latest value, sent once the window closes.
handler.set_property_debounced("motor_speed", 72.0.into(), 50).await;Typed values Properties support bool, string, integers (i8 through i64, auto-sized to the smallest fit), and f64 floats. The binary wire format encodes values with type tags for compact transmission.
Reactive streams Every property exposes a tokio::sync::watch-backed stream. Multiple consumers can independently observe changes:
let mut stream = prop.stream.clone();
tokio::spawn(async move {
while let Some(value) = stream.next().await {
// Each clone gets independent cursor
}
});Liveness detection (wave_timeout_ms) By default a Hive instance only notices a peer is gone when the OS reports the TCP connection closed. A link that dies silently, Wi-Fi drops, a process hangs instead of exiting, never triggers that. Setting wave_timeout_ms makes that Hive instance actively watch its peers instead: once a peer's gone that many milliseconds without sending anything at all, it gets pinged and given the same interval again to reply. No reply, a PONG or literally any other message both count, means it's declared dead and removed, same as a real disconnect.
name = "flight_computer"
connect = "192.168.1.10:3000"
wave_timeout_ms = 800 # ping after 800ms idle, dead if no reply within another 800msOnly the side that needs to detect a disconnect has to set this, replying to a PING is unconditional on the receiving end regardless of its own config. Watch for it in application code via handler.is_connected(). Applies to TCP and Bluetooth-peripheral peers; WebSocket and Bluetooth-central peers already emit a disconnect from their own read loops and don't need this.
is_connected() vs peer_count() is_connected() answers "is this Hive up", which for a server means "has it bound its listen port" and is true with nobody attached. peer_count() returns how many peers are actually attached, so that's the one to check for "is anyone on the other end", and the one to build a connection indicator from.
Reconnecting A Hive doesn't reconnect on its own. go() consumes it, and once its run loop is up, a peer that goes away stays away. To reconnect, cancel the CancellationToken you passed to go() and build a fresh Hive. Cancelling is honored while it's dialing too, so an attempt that hasn't connected yet can be abandoned without leaving a task retrying in the background, which is what lets a caller drive its own retry schedule instead of the built-in 10s one.
Note that a server sends its entire property set to a client as soon as that client connects. A reconnecting client therefore receives current values immediately, before anything has changed. That's usually what you want for state sync, but if a property drives something physical, treat the opening snapshot accordingly rather than as a fresh command.
Message size limit (max_message_size_bytes) Every TCP message is length-prefixed, and that length is trusted before the payload is read into memory. Without a cap, a corrupted stream or a bug on the sending side claiming an absurd size would attempt one huge allocation. Defaults to 1 MiB, comfortably more than typical property payloads need, override it only if you have a legitimate reason to send larger messages.
max_message_size_bytes = 4194304 # 4 MiB# Default, TCP only
cargo build
# With MQTT support
cargo build --features mqtt
# With WebSocket support
cargo build --features websock
# With Bluetooth support (Linux)
cargo build --features bluetooth
# Multiple features
cargo build --features "mqtt,websock"# Core tests (no external dependencies)
cargo test
# MQTT tests (requires broker on localhost:1883)
mosquitto -c mosquitto.conf -v # in another terminal
cargo test --features mqtt
# Skip MQTT tests in CI
SKIP_MQTT_TESTS=1 cargo test --features mqtt# Basic property usage
cargo run --example properties
# Run a Hive server from a TOML file
cargo run --example just_run -- examples/listen_3000.toml
# Hive with MQTT
cargo run --example mqtt_example --features mqttHive builds as a static library, cdylib, and standard Rust lib, designed to embed into iOS, Android, and Linux applications:
[lib]
crate-type = ["staticlib", "cdylib", "lib"]Android JNI bindings and a C FFI entry point (newHive) are included for native mobile integration.
┌──────────┐ TCP ┌──────────┐ TCP ┌──────────┐
│ Hive A │◄────────────►│ Hive B │◄────────────►│ Hive C │
│ (server) │ │ (middle) │ │ (client) │
└──────────┘ └──────────┘ └──────────┘
▲
│ MQTT
▼
┌──────────────┐
│ MQTT Broker │
└──────┬───────┘
│
┌──────┴───────┐
│ Hive D │
│ (MQTT only) │
└──────────────┘
Hive instances can simultaneously use multiple transports. A node can listen for TCP clients, connect to an MQTT broker, and advertise over Bluetooth, all properties stay synchronized across every transport.
MIT: see LICENSE.txt