ton/NATSBridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

v0.6.0-dev-seperate_natsclient_smartsend #13

Merged
ton merged 7 commits from v0.6.0-dev-seperate_natsclient_smartsend into v0.6.0-dev 2026-05-14 06:21:29 +00:00
Conversation 0 Commits 7 Files Changed 12 +3647 -48
11 changed files with 3647 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit c5a70edd57 - Show all commits

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
node_modules/
package.json
package-lock.json
target/

21
AI_prompt.md
View File

@@ -153,19 +153,19 @@ I'll do the other docs not listed here later myself.
now help me update the following fileaccording to ASG_Framework/ASG_Framework.md:
now help me update the following file according to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/specification.md
<!-- ------------------------------------------- 100 ------------------------------------------- -->
Check NATSBridge/docs folder. I would like to expand this package to include Dart support.
Can you update the content of the following files according to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/requirements.md
- NATSBridge/docs/specification.md
- NATSBridge/docs/walkthrough.md
- NATSBridge/docs/architecture.md
Check ./docs folder. I would like to expand this package (NATSBRIDGE) to include Rust support.
Can you update the content of the following files according to /home/ton/docker-apps/sommpanion/ASG_Framework/ASG_Framework.md:
- ./docs/requirements.md
- ./docs/specification.md
- ./docs/walkthrough.md
- ./docs/architecture.md
@@ -181,4 +181,11 @@ I updated ./src/NATSBridge.jl. Use it as groundtruth. Check ./docs folder I want
Check the following files:
- ./docs/requirements.md
- ./docs/specification.md
- ./docs/architecture.md
- ./docs/walkthrough.md
I would like to expand this package (NATSBRIDGE) to include Rust support.
Now help me update Rust implementation of this package at ./src/natsbridge.rs.

1898
Cargo.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff

31
Cargo.toml Normal file
View File

@@ -0,0 +1,31 @@
[package]
name = "natsbridge"
version = "1.2.0"
edition = "2021"
description = "Cross-platform bi-directional data bridge for NATS communication"
[lib]
name = "natsbridge"
path = "src/natsbridge.rs"
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["json", "stream"] }
uuid = { version = "1", features = ["v4", "serde"] }
base64 = "0.22"
chrono = { version = "0.4", features = ["serde"] }
async-trait = "0.1"
futures = "0.3"
[dev-dependencies]
tempfile = "3"
[[example]]
name = "smartsend_example"
path = "examples/smartsend_example.rs"
[[example]]
name = "smartreceive_example"
path = "examples/smartreceive_example.rs"

96
docs/architecture.md
View File

@@ -10,7 +10,7 @@
## 1. Executive Summary
This document defines the **blueprint** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, and **MicroPython** applications using NATS as the message bus.
This document defines the **blueprint** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This architecture document serves as the single source of truth for:
- **System Structure**: How components fit together and interact
@@ -57,6 +57,7 @@ flowchart TD
JS_App[JavaScript Application<br/>Node.js/Browser]
Python_App[Python Application<br/>Desktop]
Dart_App[Dart Application<br/>Desktop/Flutter/Web]
Rust_App[Rust Application<br/>Server/Desktop]
MicroPython_App[MicroPython Device]
end
@@ -64,12 +65,14 @@ flowchart TD
JS_App -->|NATS| NATS_Server
Python_App -->|NATS| NATS_Server
Dart_App -->|NATS| NATS_Server
Rust_App -->|NATS| NATS_Server
MicroPython_App -->|NATS| NATS_Server
Julia_App -->|HTTP| File_Server
JS_App -->|HTTP| File_Server
Python_App -->|HTTP| File_Server
Dart_App -->|HTTP| File_Server
Rust_App -->|HTTP| File_Server
MicroPython_App -->|HTTP| File_Server
style NATS_Server fill:#fff3e0,stroke:#f57c00
@@ -78,6 +81,7 @@ flowchart TD
style JS_App fill:#e3f2fd,stroke:#2196f3
style Python_App fill:#e3f2fd,stroke:#2196f3
style Dart_App fill:#fff0f6,stroke:#e91e63
style Rust_App fill:#dea584,stroke:#e65100
style MicroPython_App fill:#fce4ec,stroke:#e91e63
```
@@ -90,23 +94,15 @@ flowchart TD
JS_Module[JavaScript NATSBridge Module]
Python_Module[Python NATSBridge Module]
Dart_Module[Dart NATSBridge Module]
Rust_Module[Rust NATSBridge Module]
MicroPython_Module[MicroPython NATSBridge Module]
end
subgraph "NATS Container"
NATS_Client[NATS Client]
NATS_Broker[NATS Broker]
end
subgraph "File Server Container"
File_Client[HTTP Client]
File_Server[File Server]
end
Julia_Module --> NATS_Client
JS_Module --> NATS_Client
Python_Module --> NATS_Client
Dart_Module --> NATS_Client
Rust_Module --> NATS_Client
MicroPython_Module --> NATS_Client
NATS_Client --> NATS_Broker
@@ -115,6 +111,7 @@ flowchart TD
JS_Module --> File_Client
Python_Module --> File_Client
Dart_Module --> File_Client
Rust_Module --> File_Client
MicroPython_Module --> File_Client
File_Client --> File_Server
@@ -123,6 +120,7 @@ flowchart TD
style JS_Module fill:#e3f2fd,stroke:#2196f3
style Python_Module fill:#e3f2fd,stroke:#2196f3
style Dart_Module fill:#fff0f6,stroke:#e91e63
style Rust_Module fill:#dea584,stroke:#e65100
style MicroPython_Module fill:#fce4ec,stroke:#e91e63
style NATS_Broker fill:#fff3e0,stroke:#f57c00
style File_Server fill:#f3e5f5,stroke:#9c27b4
@@ -182,8 +180,8 @@ flowchart TD
| **_deserialize_data** | Deserialize bytes to native data types | All |
| **envelope_to_json** | Convert msg_envelope_v1 struct to JSON string | All |
| **log_trace** | Log trace messages with correlation ID | All |
| **fileserver_upload_handler** | Upload large payloads to HTTP server | Desktop (Julia/JS/Python/Dart) |
| **fileserver_download_handler** | Download payloads from HTTP server with exponential backoff | Desktop (Julia/JS/Python/Dart) |
| **fileserver_upload_handler** | Upload large payloads to HTTP server | Desktop (Julia/JS/Python/Dart/Rust) |
| **fileserver_download_handler** | Download payloads from HTTP server with exponential backoff | Desktop (Julia/JS/Python/Dart/Rust) |
### Data Flow
@@ -298,7 +296,7 @@ end
|------|-------------|---------------|----------|-----------|
| `text` | Plain text string | UTF-8 bytes | Base64 | All |
| `dictionary` | JSON object | JSON string | Base64/JSON | All |
| `arrowtable` | Apache Arrow IPC | Arrow IPC stream | Base64/arrow-ipc | Desktop (Julia/Python/Node.js/Dart) |
| `arrowtable` | Apache Arrow IPC | Arrow IPC stream | Base64/arrow-ipc | Desktop (Julia/Python/Node.js/Dart/Rust) |
| `jsontable` | JSON array of objects | JSON string | Base64/json | All (including Browser/Dart Web) |
| `image` | Binary image data | Raw bytes | Base64 | All |
| `audio` | Binary audio data | Raw bytes | Base64 | All |
@@ -535,6 +533,62 @@ DEFAULT_SIZE_THRESHOLD = 100_000 # 100KB
MAX_PAYLOAD_SIZE = 50_000 # 50KB hard limit
```
### Rust Architecture
Rust leverages compile-time type safety and async runtimes:
- **Type-safe payloads**: Rust enum discriminates between `Text`, `Dictionary`, `ArrowTable`, `Binary`, etc.
- **serde serialization**: Automatic JSON deserialization via `#[derive(Serialize, Deserialize)]`
- **tokio runtime**: Efficient async I/O for NATS connections and HTTP file server operations
- **arrow2 integration**: Native Arrow IPC deserialization without intermediate format conversion
- **reqwest**: High-performance HTTP client with built-in TLS and connection pooling
- **Zero-copy patterns**: `Vec<u8>` passed directly to avoid unnecessary memory copies
- **Result<T, E>**: Idiomatic error handling with typed error types
```rust
// Type-safe payload enum (compile-time discrimination)
#[derive(Serialize, Deserialize, Clone)]
pub enum Payload {
Text(String),
Dictionary(serde_json::Value),
ArrowTable(Vec<u8>),
JsonTable(serde_json::Value),
Image(Vec<u8>),
Audio(Vec<u8>),
Video(Vec<u8>),
Binary(Vec<u8>),
}
// Configuration via builder pattern
pub struct SmartsendOptions {
pub broker_url: String,
pub fileserver_url: String,
pub fileserver_upload_handler: Option<UploadHandler>,
pub size_threshold: usize,
pub correlation_id: String,
pub msg_purpose: String,
pub sender_name: String,
// ... other fields
}
// NATS client with tokio integration
let conn = nats::connect("nats://localhost:4222").await?;
// Subscribe and process messages
let mut sub = conn.subscribe("/agent/wine/api/v1/analyze")?;
for msg in sub.messages() {
let envelope: MsgEnvelopeV1 = serde_json::from_slice(&msg.payload)?;
// Type-safe access to payloads
for payload in &envelope.payloads {
match &payload.data {
Payload::ArrowTable(bytes) => { /* process */ },
Payload::Text(text) => { /* process */ },
_ => {}
}
}
}
```
---
## Scaling Architecture
@@ -771,7 +825,7 @@ flowchart TD
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, MicroPython 1.19+ |
---
@@ -779,6 +833,10 @@ flowchart TD
| Date | Version | Changes |
|------|---------|---------|
| 2026-05-13 | 1.3.0 | Added Rust support with tokio, serde, and arrow2 | All sections |
| - | - | Added Rust to C4 diagrams (context, container) | All sections |
| - | - | Added Rust platform-specific architecture section | specification.md:13 |
| - | - | Updated component table with Rust support | All sections |
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) |
| - | - | Removed publish_message component (commented out in source) |
| - | - | Removed NATSClient and NATSConnectionPool classes (not in ground truth) |
@@ -816,6 +874,7 @@ flowchart TD
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | specification.md:2-19 (all sections) | FR-005, FR-006, FR-012 |
### 16.3 External Dependencies
@@ -838,6 +897,13 @@ flowchart TD
| Dart | http | Latest | HTTP file server | specification.md:11 | FR-008, FR-009 |
| Dart | uuid | Latest | UUID generation | specification.md:11 | FR-011, NFR-401 |
| Dart | dart-arrow | Latest | Arrow IPC support | specification.md:11 | FR-002, FR-012 |
| Rust | nats | Latest | NATS client | specification.md:11 | FR-013, FR-014 |
| Rust | serde | Latest | JSON serialization | specification.md:11 | FR-012, NFR-101, NFR-102 |
| Rust | serde_json | Latest | JSON handling | specification.md:11 | FR-012, NFR-101, NFR-102 |
| Rust | tokio | Latest | Async runtime | specification.md:11 | FR-013, FR-014 |
| Rust | reqwest | Latest | HTTP file server | specification.md:11 | FR-008, FR-009 |
| Rust | uuid | Latest | UUID generation | specification.md:11 | FR-011, NFR-401 |
| Rust | arrow2 | Latest | Arrow IPC support | specification.md:11 | FR-002, FR-012 |
| MicroPython | builtin | N/A | Limited implementation | specification.md:11 | FR-005, FR-006, FR-012 |
---

32
docs/requirements.md
View File

@@ -11,7 +11,7 @@
### 1.1 Business Goal
NATSBridge is a cross-platform, bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, and **MicroPython** applications using NATS as the message bus. The system implements the **Claim-Check pattern** for efficient handling of large payloads (>0.5MB) by uploading them to an HTTP file server instead of sending raw binary data over NATS.
NATSBridge is a cross-platform, bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus. The system implements the **Claim-Check pattern** for efficient handling of large payloads (>0.5MB) by uploading them to an HTTP file server instead of sending raw binary data over NATS.
### 1.2 User Stories (with acceptance criteria)
@@ -25,6 +25,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| **As a Dart developer**, I want to send tabular data (List<Map>) to other platforms | P1 | JSON table format exchange works with Arrow IPC on desktop |
| **As a Dart developer**, I want to send large files (>0.5MB) | P1 | Large files are automatically uploaded to file server and URLs are sent via NATS |
| **As a MicroPython developer**, I want to send sensor data with minimal memory usage | P1 | Direct transport works for payloads <100KB on memory-constrained devices |
| **As a Rust developer**, I want to send and receive messages with type-safe APIs | P1 | Rust implementation uses serde for serialization, tokio for async, and nats-io for NATS connectivity |
| **As a developer**, I want to send mixed-content messages (text + image + file) | P1 | NATSBridge accepts list of (dataname, data, type) tuples and handles each payload appropriately |
| **As a developer**, I want to receive multi-payload messages | P1 | NATSBridge returns payloads as list of tuples with correct types preserved |
| **As a developer**, I want to use Plik as the file server | P2 | Plik one-shot upload mode is supported with upload ID and token handling |
@@ -51,7 +52,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Feature | Description |
|---------|-------------|
| Cross-platform interoperability | Seamless data exchange between Julia, JavaScript, Python, Dart, and MicroPython |
| Cross-platform interoperability | Seamless data exchange between Julia, JavaScript, Python, Dart, Rust, and MicroPython |
| Intelligent transport selection | Direct transport (<0.5MB) vs Link transport (≥0.5MB) based on payload size |
| Unified API | Consistent `smartsend()` and `smartreceive()` functions across all platforms |
| Multi-payload support | List of (dataname, data, type) tuples with appropriate handling |
@@ -88,6 +89,11 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart | nats | Latest stable |
| Dart | http | Latest stable |
| Dart | uuid | Latest stable |
| Rust | nats | Latest stable |
| Rust | serde | Latest stable |
| Rust | serde_json | Latest stable |
| Rust | tokio | Latest stable |
| Rust | uuid | Latest stable |
### 2.4 Platform Compatibility
@@ -98,6 +104,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Python | 3.8+ | pyarrow required for arrowtable support |
| Browser | Latest | No Arrow IPC (uses jsontable only) |
| Dart | 2.17+ | Supports Desktop (Dart SDK), Flutter (Dart SDK), and Web (Dart SDK) |
| Rust | 1.70+ | Full support with async/await, Arrow IPC on desktop |
| MicroPython | 1.19+ | Limited to direct transport |
---
@@ -189,14 +196,14 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Type | Julia | JavaScript | Python | Dart | MicroPython | Description |
|------|-------|------------|--------|------|-------------|-------------|
| `text` | `String` | `string` | `str` | `String` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `Map` | `dict` | JSON-serializable data |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | `List<Map>` (Desktop), `List<dynamic>` (Flutter) | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | `List<Map>` | ⚠️ | Tabular data (JSON) - **Only table type in Browser** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Image binary data |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Audio binary data |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Video binary data |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `Uint8List` | `bytearray` | Generic binary data |
| `text` | `String` | `string` | `str` | `String` | `String` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `Map`, `serde_json::Value` | `String` | `dict` | JSON-serializable data |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | `List<Map>` (Desktop), `List<dynamic>` (Flutter) | `arrow2::Table` | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | `Vec<Map>` | ⚠️ | Tabular data (JSON) - **Only table type in Browser** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Image binary data |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Audio binary data |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Video binary data |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `Uint8List` | `Vec<u8>` | `bytearray` | Generic binary data |
### 6.2 Encoding Requirements
@@ -220,6 +227,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart Desktop | 0.5MB | Default size threshold |
| Dart Flutter | 0.5MB | Default size threshold |
| Dart Web | 0.5MB | Default size threshold |
| Rust | 0.5MB | Default size threshold |
| MicroPython | 100KB | Lower threshold for memory constraints |
### 7.2 Maximum Payload Size
@@ -230,6 +238,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart Desktop | Unlimited | Limited by NATS server configuration |
| Dart Flutter | Unlimited | Limited by NATS server configuration |
| Dart Web | Unlimited | Limited by NATS server configuration |
| Rust | Unlimited | Limited by NATS server configuration |
| MicroPython | 50KB | Hard limit due to 256KB-1MB memory |
---
@@ -392,7 +401,7 @@ function smartreceive(
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Browser (latest), MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, Browser (latest), MicroPython 1.19+ |
---
@@ -419,6 +428,7 @@ function smartreceive(
- [`src/natsbridge.py`](../src/natsbridge.py) - Python implementation
- [`src/natsbridge.dart`](../src/natsbridge.dart) - Dart implementation
- [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) - MicroPython implementation
- [`src/natsbridge.rs`](../src/natsbridge.rs) - Rust implementation
- [`README.md`](../README.md) - Project overview
- [`docs/specification.md`](./specification.md) - Technical specification
- [`docs/ui-specification.md`](./ui-specification.md) - UI specification

135
docs/specification.md
View File

@@ -10,7 +10,7 @@
## 1. Technical Contract Overview
This document defines the **technical contract** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, and **MicroPython** applications using NATS as the message bus.
This document defines the **technical contract** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This specification serves as the single source of truth for:
- **Inputs**: What data structures are accepted by `smartsend()`
@@ -576,6 +576,67 @@ Future<[Map<String, dynamic>, String]> smartsend(
}
```
#### Rust
```rust
pub async fn smartsend(
subject: &str,
data: &[(String, Payload, String)],
options: &SmartsendOptions,
) -> Result<(MsgEnvelopeV1, String), NatSBridgeError>
// SmartsendOptions struct
pub struct SmartsendOptions {
pub broker_url: String,
pub fileserver_url: String,
pub fileserver_upload_handler: Option<UploadHandler>,
pub size_threshold: usize,
pub correlation_id: String,
pub msg_purpose: String,
pub sender_name: String,
pub receiver_name: String,
pub receiver_id: String,
pub reply_to: String,
pub reply_to_msg_id: String,
pub msg_id: String,
pub sender_id: String,
}
// Payload enum for type-safe data handling
#[derive(Serialize, Deserialize, Clone)]
pub enum Payload {
Text(String),
Dictionary(serde_json::Value),
ArrowTable(Vec<u8>),
JsonTable(serde_json::Value),
Image(Vec<u8>),
Audio(Vec<u8>),
Video(Vec<u8>),
Binary(Vec<u8>),
}
// MsgEnvelopeV1 struct (serde-serializable)
#[derive(Serialize, Deserialize, Clone)]
pub struct MsgEnvelopeV1 {
pub correlation_id: String,
pub msg_id: String,
pub timestamp: String,
pub send_to: String,
pub msg_purpose: String,
pub sender_name: String,
pub sender_id: String,
pub receiver_name: String,
pub receiver_id: String,
pub reply_to: String,
pub reply_to_msg_id: String,
pub broker_url: String,
pub metadata: serde_json::Value,
pub payloads: Vec<MsgPayloadV1>,
}
```
**Note**: NATS publishing is the caller's responsibility. Returns `Result<(MsgEnvelopeV1, String), NatSBridgeError>`. Uses `serde` for JSON serialization.
### `smartreceive` Function Signature
#### Julia
@@ -674,6 +735,25 @@ Future<Map<String, dynamic>> smartreceive(
}
```
#### Rust
```rust
pub async fn smartreceive(
msg_json_str: &str, // JSON string from NATS message payload
options: &SmartreceiveOptions,
) -> Result<MsgEnvelopeV1, NatSBridgeError>
// SmartreceiveOptions struct
pub struct SmartreceiveOptions {
pub fileserver_download_handler: Option<DownloadHandler>,
pub max_retries: u32,
pub base_delay: u64,
pub max_delay: u64,
}
```
**Note**: Input is JSON string from NATS message payload. Returns `Result<MsgEnvelopeV1, NatSBridgeError>`.
---
## File Server Interface
@@ -786,6 +866,18 @@ function fileserver_download_handler(
| File server download | ✅ Supported | HTTP/HTTPS |
| Size threshold | 500KB | Configurable |
### Rust
| Feature | Status | Notes |
|---------|--------|-------|
| Arrow IPC | ✅ Supported | Requires `arrow2` crate |
| JSON table | ✅ Supported | Uses `serde_json` |
| File server upload | ✅ Supported | HTTP/HTTPS via `reqwest` |
| File server download | ✅ Supported | HTTP/HTTPS via `reqwest` with retry |
| Size threshold | 500KB | Configurable |
| Async runtime | ✅ Supported | Uses `tokio` for async I/O |
| Type safety | ✅ Supported | Compile-time type checking via Rust enums |
### MicroPython
| Feature | Status | Notes |
@@ -808,6 +900,7 @@ function fileserver_download_handler(
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | Client-side rendering |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | Desktop Python |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | Desktop/Flutter/Web |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | Uses tokio + serde + arrow2 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | Memory-constrained |
### Browser Implementation Notes
@@ -822,16 +915,16 @@ The browser implementation ([`src/natsbridge_csr.js`](../src/natsbridge_csr.js))
### Payload Type Availability by Platform
| Payload Type | Julia | Node.js | Browser | Python | Dart | MicroPython |
|--------------|-------|---------|---------|--------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| Payload Type | Julia | Node.js | Browser | Python | Dart | Rust | MicroPython |
|--------------|-------|---------|---------|--------|------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
---
@@ -969,6 +1062,13 @@ flowchart TD
| Dart | http | Latest | HTTP file server |
| Dart | uuid | Latest | UUID generation |
| Dart | dart-arrow | Latest | Arrow IPC support (Desktop/Flutter) |
| Rust | nats | Latest | NATS client |
| Rust | serde | Latest | JSON serialization |
| Rust | serde_json | Latest | JSON handling |
| Rust | tokio | Latest | Async runtime |
| Rust | reqwest | Latest | HTTP file server |
| Rust | uuid | Latest | UUID generation |
| Rust | arrow2 | Latest | Arrow IPC support |
| MicroPython | builtin | N/A | Limited implementation |
### Optional Dependencies
@@ -1021,6 +1121,8 @@ flowchart TD
| [`src/natsbridge_ssr.js`](../src/natsbridge_ssr.js) | Node.js | Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | FR-005, FR-006, FR-012 |
### 20.3 External Dependencies
@@ -1039,6 +1141,17 @@ flowchart TD
| Python | nats-py | Latest | NATS client | FR-013, FR-014 |
| Python | aiohttp | Latest | HTTP file server | FR-008, FR-009 |
| Python | pyarrow | Latest | Arrow IPC support | FR-002, FR-012 |
| Dart | nats | Latest | NATS client | FR-013, FR-014 |
| Dart | http | Latest | HTTP file server | FR-008, FR-009 |
| Dart | uuid | Latest | UUID generation | FR-011, NFR-401 |
| Dart | dart-arrow | Latest | Arrow IPC support | FR-002, FR-012 |
| Rust | nats | Latest | NATS client | FR-013, FR-014 |
| Rust | serde | Latest | JSON serialization | FR-012, NFR-101, NFR-102 |
| Rust | serde_json | Latest | JSON handling | FR-012, NFR-101, NFR-102 |
| Rust | tokio | Latest | Async runtime | FR-013, FR-014 |
| Rust | reqwest | Latest | HTTP file server | FR-008, FR-009 |
| Rust | uuid | Latest | UUID generation | FR-011, NFR-401 |
| Rust | arrow2 | Latest | Arrow IPC support | FR-002, FR-012 |
| MicroPython | builtin | N/A | Limited implementation | FR-005, FR-006 |
---

158
docs/walkthrough.md
View File

@@ -9,7 +9,7 @@
## 1. Executive Summary
This document provides the **end-to-end trace** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, and **MicroPython** applications using NATS as the message bus.
This document provides the **end-to-end trace** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This walkthrough serves as the primary onboarding guide for new developers and explains:
- **User scenarios** - Real-world use cases from developer perspective
@@ -463,7 +463,155 @@ env, msg_json = smartsend(
---
## User Scenario 4: MicroPython Device
## User Scenario 4: Rust Service with Type-Safe API
### Scenario Description
A Rust service needs to process messages from a Julia analytics pipeline and send typed results back. The Rust implementation leverages compile-time type safety via Rust enums and serde for serialization.
### Step-by-Step Flow
#### Step 1: Rust Service Receives Message
```rust
// Rust service - using tokio async runtime
use natsbridge::{smartreceive, MsgEnvelopeV1};
#[tokio::main]
async fn main() {
let conn = nats::connect("nats://localhost:4222").unwrap();
// Subscribe and receive messages
let mut sub = conn.subscribe("/agent/wine/api/v1/analyze").unwrap();
for msg in sub.messages() {
let envelope: MsgEnvelopeV1 = smartreceive(
&String::from_utf8_lossy(&msg.payload),
&Default::default(),
).await.unwrap();
// Type-safe payload access
for payload in &envelope.payloads {
match &payload.data {
Payload::ArrowTable(arrow_bytes) => {
// Process Arrow IPC data using arrow2
let table = arrow2::io::ipc::read::Reader::new(
std::io::Cursor::new(arrow_bytes.clone()),
);
println!("Received {} rows", table.len());
},
Payload::Text(text) => {
println!("Message: {}", text);
},
_ => println!("Received {} bytes of {} data",
match &payload.data {
Payload::Binary(b) => b.len(),
_ => 0,
},
payload.payload_type),
}
}
}
}
```
**Rationale**:
- **Type-safe payloads**: Rust enum discriminates between payload types at compile time
- **serde serialization**: Automatic JSON deserialization to `MsgEnvelopeV1`
- **tokio runtime**: Efficient async I/O for NATS and HTTP operations
- **arrow2 integration**: Direct Arrow IPC deserialization without intermediate format
#### Step 2: Rust Service Sends Processed Results
```rust
// Rust service sends results back with mixed payload types
use natsbridge::{smartsend, Payload, SmartsendOptions};
let results_df = /* processed Arrow table */;
let result_bytes = /* serialize to Arrow IPC */;
let (envelope, json_str) = smartsend(
"/agent/wine/api/v1/results",
&[
(
"results".to_string(),
Payload::ArrowTable(result_bytes),
"arrowtable".to_string(),
),
(
"summary".to_string(),
Payload::Text("Analysis complete: 1500 rows processed".to_string()),
"text".to_string(),
),
],
&SmartsendOptions {
broker_url: "nats://localhost:4222".to_string(),
reply_to: "/python/worker/v1/results".to_string(),
msg_purpose: "chat".to_string(),
..Default::default()
},
).await?;
// Caller publishes to NATS
conn.publish("/agent/wine/api/v1/results", &json_str)?;
```
**Rationale**:
- **Builder pattern**: `SmartsendOptions` provides clean configuration
- **Enum-based payloads**: Type safety prevents sending incorrect data types
- **Default options**: sensible defaults reduce boilerplate
- **Result<T, E>**: idiomatic Rust error handling
#### Step 3: Python/Julia Receives Rust Response
```python
# Python backend receives Rust response
env = await smartreceive(str(nats_msg.payload))
# env["payloads"][0] is now:
# ("results", arrow_table_data, "arrowtable")
# env["payloads"][1] is now:
# ("summary", "Analysis complete: 1500 rows processed", "text")
```
**Rationale**:
- **Cross-platform parity**: Rust envelope matches other platform envelopes exactly
- **Same JSON wire format**: No protocol translation needed
- **Type preservation**: Arrow IPC and text types preserved across all platforms
#### Step 4: Large File Transfer from Rust
```rust
// Rust service sends large binary file via link transport
let large_file_data: Vec<u8> = std::fs::read("/data/large_dataset.parquet")?;
let (envelope, json_str) = smartsend(
"/agent/wine/api/v1/upload",
&[
(
"dataset".to_string(),
Payload::Binary(large_file_data),
"binary".to_string(),
),
],
&SmartsendOptions {
broker_url: "nats://localhost:4222".to_string(),
fileserver_url: "http://localhost:8080".to_string(),
size_threshold: 500_000, // 0.5MB triggers link transport
..Default::default()
},
).await?;
```
**Rationale**:
- **Automatic transport selection**: Same 0.5MB threshold as other desktop platforms
- **reqwest integration**: Efficient HTTP client for file server upload/download
- **Exponential backoff**: Built-in retry with configurable parameters
- **Zero-copy where possible**: `Vec<u8>` passed directly without intermediate copies
---
## User Scenario 5: MicroPython Device
### Scenario Description
@@ -528,7 +676,7 @@ env = await smartreceive(str(nats_msg.payload))
---
## User Scenario 5: Cross-Platform Chat with Mixed Payloads
## User Scenario 6: Cross-Platform Chat with Mixed Payloads
### Scenario Description
@@ -763,6 +911,7 @@ log_trace(correlation_id, "Published to NATS")
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | specification.md:2-19 (all sections) |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | specification.md:2-19 (all sections) |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | specification.md:2-19 (all sections) |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | specification.md:2-19 (all sections) |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | specification.md:2-19 (all sections) |
---
@@ -771,6 +920,9 @@ log_trace(correlation_id, "Published to NATS")
| Date | Version | Changes | Specification Reference |
|------|---------|---------|------------------------|
| 2026-05-13 | 1.3.0 | Added Rust support with tokio, serde, and arrow2 | All sections |
| - | - | Added Rust user scenario (User Scenario 4) | specification.md:11 (Rust API) |
| - | - | Updated scenario numbering (MicroPython → Scenario 5, Cross-Platform → Scenario 6) | All sections |
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) | All sections |
| - | - | Updated smartreceive calls to use String(nats_msg.payload) pattern | All sections |
| - | - | Removed NATSClient.publish() calls (caller responsible for NATS publishing) | All sections |

96
examples/smartreceive_example.rs Normal file
View File

@@ -0,0 +1,96 @@
use natsbridge::{smartreceive, SmartreceiveOptions};
#[tokio::main]
async fn main() {
// Simulated NATS message JSON (received from NATS subscription)
let msg_json_str = r#"{
"correlation_id": "abc123-def456-ghi789",
"msg_id": "msg-uuid-001",
"timestamp": "2026-05-13T12:00:00.000Z",
"send_to": "/agent/wine/api/v1/prompt",
"msg_purpose": "chat",
"sender_name": "js-webapp",
"sender_id": "sender-uuid-001",
"receiver_name": "rust-backend",
"receiver_id": "",
"reply_to": "/agent/wine/api/v1/response",
"reply_to_msg_id": "",
"broker_url": "nats://localhost:4222",
"metadata": {},
"payloads": [
{
"id": "payload-uuid-001",
"dataname": "message",
"payload_type": "text",
"transport": "direct",
"encoding": "base64",
"size": 29,
"data": "SGVsbG8gZnJvbSBKYXZhU2NyaXB0ISE=",
"metadata": {"payload_bytes": 29}
},
{
"id": "payload-uuid-002",
"dataname": "user_data",
"payload_type": "dictionary",
"transport": "direct",
"encoding": "json",
"size": 58,
"data": "eyJ0eXBlIjoiY2hhdCIsInNlbmRlciI6InNlcnZpY2VBIiwicmVjZWl2ZXIiOiJzZXJ2aWNlQiJ9",
"metadata": {"payload_bytes": 58}
}
]
}"#;
let options = SmartreceiveOptions::default();
match smartreceive(msg_json_str, &options).await {
Ok(envelope) => {
println!("=== Envelope Received ===");
println!("Correlation ID: {}", envelope.correlation_id);
println!("Message ID: {}", envelope.msg_id);
println!("Subject: {}", envelope.send_to);
println!("Purpose: {}", envelope.msg_purpose);
println!("Sender: {}", envelope.sender_name);
println!("Receiver: {}", envelope.receiver_name);
println!("Payloads: {}", envelope.payloads.len());
println!();
for payload in &envelope.payloads {
println!("--- Payload: {} ---", payload.dataname);
println!(" Type: {}", payload.payload_type);
println!(" Transport: {}", payload.transport);
println!(" Encoding: {}", payload.encoding);
println!(" Size: {} bytes", payload.size);
// In a real scenario, you would deserialize payload.data here
// based on payload_type to get the actual data
match payload.payload_type.as_str() {
"text" => {
// For demonstration, decode the base64
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
if payload.transport == "direct" {
let decoded = BASE64.decode(&payload.data).unwrap();
println!(" Data: {}", String::from_utf8_lossy(&decoded));
} else {
println!(" URL: {}", payload.data);
}
}
"dictionary" => {
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
if payload.transport == "direct" {
let decoded = BASE64.decode(&payload.data).unwrap();
let json: serde_json::Value = serde_json::from_slice(&decoded).unwrap();
println!(" Data: {}", serde_json::to_string_pretty(&json).unwrap());
}
}
other => {
println!(" Data type: {}", other);
}
}
}
}
Err(e) => {
eprintln!("Error: {}", e);
}
}
}

70
examples/smartsend_example.rs Normal file
View File

@@ -0,0 +1,70 @@
use natsbridge::{smartsend, Payload, SmartsendOptions};
#[tokio::main]
async fn main() {
// Create mixed payload data
let payloads = vec![
(
"message".to_string(),
Payload::Text("Hello from Rust!".to_string()),
"text".to_string(),
),
(
"user_data".to_string(),
Payload::Dictionary(serde_json::json!({
"name": "Alice",
"role": "admin",
"scores": [95, 88, 92]
})),
"dictionary".to_string(),
),
(
"avatar".to_string(),
Payload::Binary(vec![0x89, 0x50, 0x4E, 0x47]), // PNG header
"image".to_string(),
),
];
let options = SmartsendOptions {
broker_url: "nats://localhost:4222".to_string(),
fileserver_url: "http://localhost:8080".to_string(),
msg_purpose: "chat".to_string(),
sender_name: "rust-example".to_string(),
..Default::default()
};
match smartsend("/agent/wine/api/v1/prompt", &payloads, &options).await {
Ok((envelope, json_str)) => {
println!("=== Envelope Created ===");
println!("Correlation ID: {}", envelope.correlation_id);
println!("Message ID: {}", envelope.msg_id);
println!("Timestamp: {}", envelope.timestamp);
println!("Subject: {}", envelope.send_to);
println!("Purpose: {}", envelope.msg_purpose);
println!("Sender: {}", envelope.sender_name);
println!("Payloads: {}", envelope.payloads.len());
println!();
for payload in &envelope.payloads {
println!("Payload: {} (type: {}, transport: {}, encoding: {})",
payload.dataname,
payload.payload_type,
payload.transport,
payload.encoding);
println!(" Size: {} bytes", payload.size);
println!(" Data: {}", if payload.transport == "direct" {
&payload.data[..payload.data.len().min(40)]
} else {
&payload.data[..payload.data.len().min(60)]
});
}
println!();
println!("=== JSON String for NATS Publishing ===");
println!("{}", json_str);
}
Err(e) => {
eprintln!("Error: {}", e);
}
}
}

1155
src/natsbridge.rs Normal file
View File

File diff suppressed because it is too large Load Diff