> ## Documentation Index
> Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
> Use this file to discover all available pages before exploring further.
# Streaming
> Stream operations for atomic data updates
## Overview
The Streams API provides atomic operations on stream data. Streams are key-value stores that support atomic updates with multiple operations (increment, set, merge, etc.).
## Creating a Streams Instance
### Streams::new
Create a new Streams instance for performing atomic updates.
```rust theme={null}
pub fn new(iii: III) -> Self
```
III client instance
**Example:**
```rust theme={null}
use iii_sdk::{III, Streams};
let iii = III::new("ws://localhost:49134");
iii.connect().await?;
let streams = Streams::new(iii);
```
## Stream Keys
Stream keys follow the format: `stream_name::group_id::item_id`
* `stream_name`: The stream namespace
* `group_id`: Group identifier (e.g., user ID, session ID)
* `item_id`: Item identifier within the group
**Examples:**
* `orders::user-123::order-456`
* `counters::daily::page-views`
* `sessions::session-abc::state`
## Atomic Operations
### update
Perform atomic updates with multiple operations.
```rust theme={null}
pub async fn update(
&self,
key: impl Into,
ops: Vec,
) -> Result
```
Stream key in the format `stream::group::item`
List of operations to apply atomically
Result containing old and new values
**Example:**
```rust theme={null}
use iii_sdk::{Streams, UpdateOp};
use serde_json::json;
let streams = Streams::new(iii);
let result = streams.update(
"orders::user-123::order-456",
vec![
UpdateOp::increment("total", 100),
UpdateOp::set("status", json!("processing")),
UpdateOp::set("updated_at", json!("2024-01-15T10:30:00Z")),
],
).await?;
println!("Old value: {:?}", result.old_value);
println!("New value: {:?}", result.new_value);
```
All operations in an `update` call are applied atomically. Either all succeed or none are applied.
## Convenience Methods
### increment
Atomically increment a numeric field.
```rust theme={null}
pub async fn increment(
&self,
key: impl Into,
field: impl Into,
by: i64,
) -> Result
```
**Example:**
```rust theme={null}
// Increment page view counter
streams.increment("counters::daily::page-views", "count", 1).await?;
// Increment by larger amount
streams.increment("wallet::user-123::balance", "amount", 1000).await?;
```
### decrement
Atomically decrement a numeric field.
```rust theme={null}
pub async fn decrement(
&self,
key: impl Into,
field: impl Into,
by: i64,
) -> Result
```
**Example:**
```rust theme={null}
// Decrement inventory
streams.decrement("inventory::warehouse-1::item-abc", "quantity", 5).await?;
```
### set\_field
Atomically set a field value.
```rust theme={null}
pub async fn set_field(
&self,
key: impl Into,
field: impl Into,
value: impl Into,
) -> Result
```
**Example:**
```rust theme={null}
use serde_json::json;
// Set user status
streams.set_field(
"users::active::user-123",
"status",
json!("online")
).await?;
// Set nested field
streams.set_field(
"profiles::user-123::data",
"settings.theme",
json!("dark")
).await?;
```
### remove\_field
Atomically remove a field.
```rust theme={null}
pub async fn remove_field(
&self,
key: impl Into,
field: impl Into,
) -> Result
```
**Example:**
```rust theme={null}
// Remove a field
streams.remove_field("cache::session-abc::data", "expired_token").await?;
```
### merge
Atomically merge an object into the existing value.
```rust theme={null}
pub async fn merge(
&self,
key: impl Into,
value: impl Into,
) -> Result
```
**Example:**
```rust theme={null}
use serde_json::json;
// Merge settings
streams.merge(
"settings::user-123::preferences",
json!({
"theme": "dark",
"language": "en",
"notifications": true
})
).await?;
```
## Update Operations
### UpdateOp
Operations that can be performed atomically.
```rust theme={null}
pub enum UpdateOp {
Set { path: FieldPath, value: Option },
Merge { path: Option, value: Value },
Increment { path: FieldPath, by: i64 },
Decrement { path: FieldPath, by: i64 },
Remove { path: FieldPath },
}
```
### Creating UpdateOps
```rust theme={null}
use iii_sdk::UpdateOp;
use serde_json::json;
// Set operation
let op = UpdateOp::set("name", json!("Alice"));
// Increment operation
let op = UpdateOp::increment("counter", 1);
// Decrement operation
let op = UpdateOp::decrement("stock", 5);
// Remove operation
let op = UpdateOp::remove("temp_data");
// Merge operation (root level)
let op = UpdateOp::merge(json!({ "new_field": "value" }));
// Merge operation (at path)
let op = UpdateOp::merge_at("settings", json!({ "theme": "dark" }));
```
## UpdateBuilder
Build complex update operations fluently.
```rust theme={null}
use iii_sdk::UpdateBuilder;
use serde_json::json;
let ops = UpdateBuilder::new()
.increment("views", 1)
.set("last_viewed", json!("2024-01-15T10:30:00Z"))
.merge(json!({
"metadata": {
"source": "web"
}
}))
.build();
let result = streams.update("analytics::page-123::stats", ops).await?;
```
## Result Types
### UpdateResult
Result of an atomic update operation.
```rust theme={null}
pub struct UpdateResult {
pub old_value: Option,
pub new_value: Value,
}
```
**Example:**
```rust theme={null}
let result = streams.increment("counter::daily::visits", "count", 1).await?;
if let Some(old) = result.old_value {
println!("Previous count: {}", old.get("count").unwrap());
}
println!("New count: {}", result.new_value.get("count").unwrap());
```
## Field Paths
Field paths use dot notation to access nested fields:
```rust theme={null}
// Top-level field
UpdateOp::set("status", json!("active"))
// Nested field
UpdateOp::set("user.profile.name", json!("Alice"))
// Array index
UpdateOp::set("items[0].quantity", json!(5))
```
## Complete Example
Here's a complete example showing stream operations:
```rust theme={null}
use iii_sdk::{III, Streams, UpdateOp, UpdateBuilder};
use serde_json::json;
#[tokio::main]
async fn main() -> Result<(), Box> {
let iii = III::new("ws://localhost:49134");
iii.connect().await?;
let streams = Streams::new(iii);
// Initialize a counter
streams.set_field(
"analytics::app::daily-stats",
"views",
json!(0)
).await?;
// Increment counter atomically
for _ in 0..10 {
streams.increment("analytics::app::daily-stats", "views", 1).await?;
}
// Update multiple fields atomically
let result = streams.update(
"analytics::app::daily-stats",
vec![
UpdateOp::increment("unique_visitors", 1),
UpdateOp::set("last_updated", json!("2024-01-15T10:30:00Z")),
UpdateOp::merge(json!({
"metadata": {
"version": "1.0",
"platform": "web"
}
})),
],
).await?;
println!("Updated stats: {:?}", result.new_value);
// Use builder for complex updates
let ops = UpdateBuilder::new()
.increment("views", 5)
.increment("clicks", 2)
.set("status", json!("active"))
.build();
streams.update("analytics::app::daily-stats", ops).await?;
Ok(())
}
```
## Use Cases
### Counters and Analytics
```rust theme={null}
// Track page views
streams.increment("analytics::page-123::stats", "views", 1).await?;
// Track unique visitors
streams.update(
"analytics::page-123::stats",
vec![
UpdateOp::increment("views", 1),
UpdateOp::increment("unique_visitors", 1),
],
).await?;
```
### User Sessions
```rust theme={null}
// Update session state
streams.merge(
"sessions::user-123::current",
json!({
"last_activity": "2024-01-15T10:30:00Z",
"page": "/dashboard",
"online": true
})
).await?;
```
### Inventory Management
```rust theme={null}
// Decrement stock atomically
let result = streams.decrement(
"inventory::warehouse-1::item-abc",
"quantity",
5
).await?;
if let Some(old) = result.old_value {
if old.get("quantity").unwrap().as_i64().unwrap() < 5 {
println!("Warning: Low stock!");
}
}
```
## See Also
* [Channels API](/rust/api/channels) - Stream binary data between workers
* [Functions API](/rust/api/functions) - Process stream updates in functions
* [Triggers API](/rust/api/triggers) - React to stream changes with triggers