2 Commits

Author SHA1 Message Date
14cd4c0076 update doc 2026-03-24 08:55:06 +07:00
5191f1aae5 update 2026-03-23 16:15:15 +07:00
18 changed files with 1641 additions and 4048 deletions

1
.gitignore vendored
View File

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

View File

@@ -143,7 +143,7 @@ Since I develop src folder before I adopt SDD_FRAMEWORK.md approach, can you che
# ---------------------------------------------- 100 --------------------------------------------- #
I updated src/NATSBridge.jl. Check and NATSBridge/docs folder I want to update the content of the following files according to ASG_Framework/ASG_Framework.md:
Check NATSBridge/docs folder I want to update the content of the following files according to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/requirements.md
- NATSBridge/docs/specification.md
- NATSBridge/docs/ui-specification.md (you'll need to create this one)
@@ -153,50 +153,20 @@ I'll do the other docs not listed here later myself.
now help me update the following file according to ASG_Framework/ASG_Framework.md:
now help me update the following fileaccording to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/specification.md
<!-- ------------------------------------------- 100 ------------------------------------------- -->
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
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
<!-- ------------------------------------------- 100 ------------------------------------------- -->
I updated ./src/NATSBridge.jl. Use it as groundtruth. Check ./docs folder I want to 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
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.
I want to build a client-side-rendering Dioxus-based chat webapp.
Dioxus version 0.7+ should be great.
I already populate the current folder for the project.
my server REST API endpoint is sommpanion.yiem.cc/agent-fronent/api/v1/chat but I didn't run the server yet. A message format is JSON string.
I just placed my custom package for encode and decode message at ./src/natsbridge.rs. smartsend() is for encoding and smartreceive() is for decoding.
you may also check the file /home/ton/docker-apps/sommpanion/NATSBridge/docs/walkthrough.md for more info about my package.
You can test whether Dioxus webapp can be build using this command "dx bundle --web --release --debug-symbols=false"

1915
Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,31 +0,0 @@
[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", "multipart"] }
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"

View File

@@ -52,9 +52,9 @@ version = "1.2.2"
[[deps.CSV]]
deps = ["CodecZlib", "Dates", "FilePathsBase", "InlineStrings", "Mmap", "Parsers", "PooledArrays", "PrecompileTools", "SentinelArrays", "Tables", "Unicode", "WeakRefStrings", "WorkerUtilities"]
git-tree-sha1 = "deddd8725e5e1cc49ee205a1964256043720a6c3"
git-tree-sha1 = "8d8e0b0f350b8e1c91420b5e64e5de774c2f0f4d"
uuid = "336ed68f-0bac-5ca0-87d4-7b16caf5d00b"
version = "0.10.15"
version = "0.10.16"
[[deps.CodeTracking]]
deps = ["InteractiveUtils", "UUIDs"]
@@ -108,9 +108,9 @@ version = "1.3.0+1"
[[deps.ConcurrentUtilities]]
deps = ["Serialization", "Sockets"]
git-tree-sha1 = "d9d26935a0bcffc87d2613ce14c527c99fc543fd"
git-tree-sha1 = "21d088c496ea22914fe80906eb5bce65755e5ec8"
uuid = "f0e56b4a-5159-44fe-b623-3e5288b988bb"
version = "2.5.0"
version = "2.5.1"
[[deps.Crayons]]
git-tree-sha1 = "249fe38abf76d48563e2f4556bebd215aa317e15"
@@ -171,9 +171,9 @@ uuid = "f43a241f-c20a-4ad4-852c-f6b1247861c6"
version = "1.7.0"
[[deps.EnumX]]
git-tree-sha1 = "7bebc8aad6ee6217c78c5ddcf7ed289d65d0263e"
git-tree-sha1 = "c49898e8438c828577f04b92fc9368c388ac783c"
uuid = "4e289a0a-7415-4d19-859d-a7e5c4648b56"
version = "1.0.6"
version = "1.0.7"
[[deps.ExceptionUnwrapping]]
deps = ["Test"]
@@ -234,9 +234,9 @@ version = "0.3.1"
[[deps.HTTP]]
deps = ["Base64", "CodecZlib", "ConcurrentUtilities", "Dates", "ExceptionUnwrapping", "Logging", "LoggingExtras", "MbedTLS", "NetworkOptions", "OpenSSL", "PrecompileTools", "Random", "SimpleBufferStream", "Sockets", "URIs", "UUIDs"]
git-tree-sha1 = "5e6fe50ae7f23d171f44e311c2960294aaa0beb5"
git-tree-sha1 = "51059d23c8bb67911a2e6fd5130229113735fc7e"
uuid = "cd3eb016-35fb-5094-929b-558a96fad6f3"
version = "1.10.19"
version = "1.11.0"
[[deps.HashArrayMappedTries]]
git-tree-sha1 = "2eaa69a7cab70a52b9687c8bf950a5a93ec895ae"
@@ -307,9 +307,9 @@ weakdeps = ["ArrowTypes"]
[[deps.JuliaInterpreter]]
deps = ["CodeTracking", "InteractiveUtils", "Random", "UUIDs"]
git-tree-sha1 = "80580012d4ed5a3e8b18c7cd86cebe4b816d17a6"
git-tree-sha1 = "3d3b79166e2a0afcf875df20db110af91ad3ab61"
uuid = "aa1ae85d-cabe-5617-a682-6adf51b2e16a"
version = "0.10.9"
version = "0.10.11"
[[deps.JuliaSyntaxHighlighting]]
deps = ["StyledStrings"]
@@ -383,9 +383,9 @@ version = "1.2.0"
[[deps.LoweredCodeUtils]]
deps = ["CodeTracking", "Compiler", "JuliaInterpreter"]
git-tree-sha1 = "65ae3db6ab0e5b1b5f217043c558d9d1d33cc88d"
git-tree-sha1 = "5d4278f755440f70648d80cc6225f51e78e94094"
uuid = "6f1432cf-f94c-5a45-995e-cdbf5db27b0b"
version = "3.5.0"
version = "3.5.1"
[[deps.Lz4_jll]]
deps = ["Artifacts", "JLLWrappers", "Libdl"]
@@ -400,9 +400,9 @@ version = "1.11.0"
[[deps.MbedTLS]]
deps = ["Dates", "MbedTLS_jll", "MozillaCACerts_jll", "NetworkOptions", "Random", "Sockets"]
git-tree-sha1 = "c067a280ddc25f196b5e7df3877c6b226d390aaf"
git-tree-sha1 = "8785729fa736197687541f7053f6d8ab7fc44f92"
uuid = "739be429-bea8-5141-9913-cc70e7f3736d"
version = "1.1.9"
version = "1.1.10"
[[deps.MbedTLS_jll]]
deps = ["Artifacts", "JLLWrappers", "Libdl"]
@@ -437,10 +437,10 @@ uuid = "55e73f9c-eeeb-467f-b4cc-a633fde63d2a"
version = "0.1.0"
[[deps.NATSBridge]]
deps = ["Arrow", "DataFrames", "Dates", "GeneralUtils", "HTTP", "JSON", "NATS", "PrettyPrinting", "Revise", "UUIDs"]
deps = ["Arrow", "Base64", "DataFrames", "Dates", "GeneralUtils", "HTTP", "JSON", "NATS", "PrettyPrinting", "Revise", "UUIDs"]
path = "."
uuid = "f2724d33-f338-4a57-b9f8-1be882570d10"
version = "0.4.1"
version = "0.5.6"
[[deps.NanoDates]]
deps = ["Dates", "Parsers"]
@@ -514,9 +514,9 @@ version = "1.3.3"
[[deps.Preferences]]
deps = ["TOML"]
git-tree-sha1 = "522f093a29b31a93e34eaea17ba055d850edea28"
git-tree-sha1 = "8b770b60760d4451834fe79dd483e318eee709c4"
uuid = "21216c6a-2e73-6563-6e65-726566657250"
version = "1.5.1"
version = "1.5.2"
[[deps.PrettyPrinting]]
git-tree-sha1 = "142ee93724a9c5d04d78df7006670a93ed1b244e"
@@ -525,9 +525,15 @@ version = "0.4.2"
[[deps.PrettyTables]]
deps = ["Crayons", "LaTeXStrings", "Markdown", "PrecompileTools", "Printf", "REPL", "Reexport", "StringManipulation", "Tables"]
git-tree-sha1 = "c5a07210bd060d6a8491b0ccdee2fa0235fc00bf"
git-tree-sha1 = "211530a7dc76ab59087f4d4d1fc3f086fbe87594"
uuid = "08abe8d2-0d0c-5749-adfa-8a2ac140af0d"
version = "3.1.2"
version = "3.2.3"
[deps.PrettyTables.extensions]
PrettyTablesTypstryExt = "Typstry"
[deps.PrettyTables.weakdeps]
Typstry = "f0ed7684-a786-439e-b1e3-3b82803b501e"
[[deps.Printf]]
deps = ["Unicode"]
@@ -535,9 +541,9 @@ uuid = "de0858da-6303-5e67-8744-51eddeeeb8d7"
version = "1.11.0"
[[deps.PtrArrays]]
git-tree-sha1 = "1d36ef11a9aaf1e8b74dacc6a731dd1de8fd493d"
git-tree-sha1 = "4fbbafbc6251b883f4d2705356f3641f3652a7fe"
uuid = "43287f4e-b6f4-7ad1-bb20-aadabca52c3d"
version = "1.3.0"
version = "1.4.0"
[[deps.QuadGK]]
deps = ["DataStructures", "LinearAlgebra"]
@@ -568,9 +574,9 @@ version = "1.2.2"
[[deps.Revise]]
deps = ["CodeTracking", "FileWatching", "InteractiveUtils", "JuliaInterpreter", "LibGit2", "LoweredCodeUtils", "OrderedCollections", "Preferences", "REPL", "UUIDs"]
git-tree-sha1 = "14d1bfb0a30317edc77e11094607ace3c800f193"
git-tree-sha1 = "d97d78d4fc5f858d8ce44f6b88bc972f2023f51d"
uuid = "295af30f-e4ad-537b-8983-00126c2a3abe"
version = "3.13.2"
version = "3.14.0"
[deps.Revise.extensions]
DistributedExt = "Distributed"
@@ -596,9 +602,9 @@ version = "0.7.0"
[[deps.ScopedValues]]
deps = ["HashArrayMappedTries", "Logging"]
git-tree-sha1 = "c3b2323466378a2ba15bea4b2f73b081e022f473"
git-tree-sha1 = "ac4b837d89a58c848e85e698e2a2514e9d59d8f6"
uuid = "7e506255-f358-4e82-b7e4-beb19740aa63"
version = "1.5.0"
version = "1.6.0"
[[deps.Scratch]]
deps = ["Dates"]
@@ -644,9 +650,9 @@ version = "1.12.0"
[[deps.SpecialFunctions]]
deps = ["IrrationalConstants", "LogExpFunctions", "OpenLibm_jll", "OpenSpecFun_jll"]
git-tree-sha1 = "f2685b435df2613e25fc10ad8c26dddb8640f547"
git-tree-sha1 = "5acc6a41b3082920f79ca3c759acbcecf18a8d78"
uuid = "276daf66-3868-5448-9aa4-cd146d93841b"
version = "2.6.1"
version = "2.7.1"
[deps.SpecialFunctions.extensions]
SpecialFunctionsChainRulesCoreExt = "ChainRulesCore"
@@ -692,9 +698,9 @@ version = "1.5.2"
[[deps.StringManipulation]]
deps = ["PrecompileTools"]
git-tree-sha1 = "a3c1536470bf8c5e02096ad4853606d7c8f62721"
git-tree-sha1 = "d05693d339e37d6ab134c5ab53c29fce5ee5d7d5"
uuid = "892a3eda-7b42-436c-8928-eab12a02cf0e"
version = "0.4.2"
version = "0.4.4"
[[deps.StringViews]]
git-tree-sha1 = "f2dcb92855b31ad92fe8f079d4f75ac57c93e4b8"
@@ -709,16 +715,18 @@ version = "1.11.0"
[[deps.StructUtils]]
deps = ["Dates", "UUIDs"]
git-tree-sha1 = "9297459be9e338e546f5c4bedb59b3b5674da7f1"
git-tree-sha1 = "fa95b3b097bcef5845c142ea2e085f1b2591e92c"
uuid = "ec057cc2-7a8d-4b58-b3b3-92acb9f63b42"
version = "2.6.2"
version = "2.7.1"
[deps.StructUtils.extensions]
StructUtilsMeasurementsExt = ["Measurements"]
StructUtilsStaticArraysCoreExt = ["StaticArraysCore"]
StructUtilsTablesExt = ["Tables"]
[deps.StructUtils.weakdeps]
Measurements = "eff96d63-e80a-5855-80a2-b1b0885c5ab7"
StaticArraysCore = "1e83bf80-4336-4d27-bf5d-d5a4f845583c"
Tables = "bd369af6-aec1-5ad0-b16a-f7cc5008161c"
[[deps.StyledStrings]]

187
README.md
View File

@@ -1,6 +1,6 @@
# NATSBridge - Cross-Platform Bi-Directional Data Bridge
A high-performance, bi-directional data bridge for **Julia**, **JavaScript**, **Python**, and **MicroPython** applications using NATS (Core & JetStream), implementing the Claim-Check pattern for large payloads.
A high-performance, bi-directional data bridge for **Julia**, **JavaScript**, **Python**, **Dart**, and **MicroPython** applications using NATS (Core & JetStream), implementing the Claim-Check pattern for large payloads.
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)
[![NATS](https://img.shields.io/badge/NATS-Enabled-green.svg)](https://nats.io)
@@ -48,33 +48,35 @@ NATSBridge enables seamless communication across multiple platforms through NATS
| **JavaScript (Node.js)** | [`src/natsbridge_ssr.js`](src/natsbridge_ssr.js) | Node.js, async/await, Arrow IPC |
| **JavaScript (Browser)** | [`src/natsbridge_csr.js`](src/natsbridge_csr.js) | Browser, WebSocket NATS, async/await, JSON table only |
| **Python** | [`src/natsbridge.py`](src/natsbridge.py) | Desktop Python, asyncio, type hints, Arrow IPC |
| **Dart (Desktop/Flutter)** | [`src/natsbridge.dart`](src/natsbridge.dart) | Desktop/Flutter, async/await, Arrow IPC |
| **Dart Web** | [`src/natsbridge.dart`](src/natsbridge.dart) | Web, WebSocket NATS, JSON table only |
| **MicroPython** | [`src/natsbridge_mpy.py`](src/natsbridge_mpy.py) | Memory-constrained, synchronous API |
### Platform Comparison
| Feature | Julia | JavaScript | JavaScript (Browser) | Python | MicroPython |
|---------|-------|------------|----------------------|--------|-------------|
| Multiple Dispatch | ✅ Native | ❌ | ❌ | ❌ | ❌ |
| Async/Await | ❌ | ✅ Native | ✅ Native | ✅ Native | ⚠️ (uasyncio) |
| Type Safety | ✅ Strong | ⚠️ (TypeScript) | ⚠️ (TypeScript) | ✅ (Type hints) | ❌ |
| Arrow IPC | ✅ Native | ✅ Native | ❌ (Browser incompatible) | ✅ Native | ❌ |
| JSON Table | ✅ | ✅ | ✅ (Only table type) | ✅ | ⚠️ (Limited) |
| Direct Transport | ✅ | ✅ | ✅ | ✅ | ✅ |
| Link Transport | ✅ | ✅ | ✅ | ✅ | ⚠️ (Limited) |
| Handler Functions | ✅ | ✅ | ✅ | ✅ | ✅ |
| Cross-Platform API | ✅ | ✅ | ✅ | ✅ | ✅ |
| WebSocket NATS | ❌ | ❌ | ✅ | ❌ | ❌ |
| Feature | Julia | JavaScript | JavaScript (Browser) | Python | Dart | Dart Web | MicroPython |
|---------|-------|------------|----------------------|--------|------|----------|-------------|
| Multiple Dispatch | ✅ Native | ❌ | ❌ | ❌ | ❌ | ❌ | ❌ |
| Async/Await | ❌ | ✅ Native | ✅ Native | ✅ Native | ✅ Native | ✅ Native | ⚠️ (uasyncio) |
| Type Safety | ✅ Strong | ⚠️ (TypeScript) | ⚠️ (TypeScript) | ✅ (Type hints) | ✅ Strong | ✅ Strong | ❌ |
| Arrow IPC | ✅ Native | ✅ Native | ❌ (Browser incompatible) | ✅ Native | ✅ Native | ❌ (Browser incompatible) | ❌ |
| JSON Table | ✅ | ✅ | ✅ (Only table type) | ✅ | ✅ | ✅ | ⚠️ (Limited) |
| Direct Transport | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| Link Transport | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ (Limited) |
| Handler Functions | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| Cross-Platform API | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| WebSocket NATS | ❌ | ❌ | ✅ | ❌ | ✅ | ✅ | ❌ |
---
## Features
-**Cross-platform messaging** for Julia, JavaScript, Python, and MicroPython applications
-**Cross-platform messaging** for Julia, JavaScript, Python, Dart, and MicroPython applications
-**Bi-directional messaging** with request-reply patterns
-**Multi-payload support** - send multiple payloads with different types in one message
-**Automatic transport selection** - direct vs link based on payload size
-**Claim-Check pattern** for payloads ≥ 500KB
-**Apache Arrow IPC** support for tabular data (Desktop: Julia/Python/Node.js)
-**Apache Arrow IPC** support for tabular data (Desktop: Julia/Python/Node.js/Dart)
-**JSON Table** support for tabular data (All platforms including Browser)
-**Exponential backoff** for reliable file server downloads
-**Correlation ID tracking** for message tracing
@@ -171,6 +173,38 @@ env, env_json_str = smartsend(
print("Message sent!")
```
#### Dart (Desktop/Flutter)
```dart
import 'package:natsbridge/natsbridge.dart';
final data = [
['message', 'Hello World', 'text']
];
final [env, envJsonStr] = await NATSBridge.send(
'/chat/room1',
data,
brokerUrl: 'nats://localhost:4222',
);
print('Message sent!');
```
#### Dart Web
```dart
import 'package:natsbridge/natsbridge.dart';
final data = [
['message', 'Hello World', 'text']
];
final [env, envJsonStr] = await NATSBridge.send(
'/chat/room1',
data,
brokerUrl: 'ws://localhost:4222', // WebSocket for browser
);
print('Message sent!');
```
---
## API Reference
@@ -335,6 +369,54 @@ env, env_json_str = NATSBridge.smartsend(
# Returns: Tuple[Dict, str]
```
#### Dart (Desktop/Flutter)
```dart
import 'package:natsbridge/natsbridge.dart';
final env, envJsonStr = await NATSBridge.send(
subject,
data, // List of [dataname, data, type] lists
brokerUrl: 'nats://localhost:4222',
fileserverUrl: 'http://localhost:8080',
fileserverUploadHandler: plikOneshotUpload,
sizeThreshold: 500000,
correlationId: uuid.v4(),
msgPurpose: 'chat',
senderName: 'NATSBridge',
receiverName: '',
receiverId: '',
replyTo: '',
replyToMsgId: '',
isPublish: true,
);
// Returns: Future<List<dynamic>> [env, env_json_str]
```
#### Dart Web
```dart
import 'package:natsbridge/natsbridge.dart';
final env, envJsonStr = await NATSBridge.send(
subject,
data, // List of [dataname, data, type] lists
brokerUrl: 'ws://localhost:4222', // WebSocket for browser
fileserverUrl: 'http://localhost:8080',
fileserverUploadHandler: plikOneshotUpload,
sizeThreshold: 500000,
correlationId: uuid.v4(),
msgPurpose: 'chat',
senderName: 'NATSBridge',
receiverName: '',
receiverId: '',
replyTo: '',
replyToMsgId: '',
isPublish: true,
);
// Returns: Future<List<dynamic>> [env, env_json_str]
```
### smartreceive
Receives and processes messages from NATS, handling both direct and link transport.
@@ -418,20 +500,50 @@ env = NATSBridge.smartreceive(
# Returns: Dict with "payloads" key
```
#### Dart (Desktop/Flutter)
```dart
import 'package:natsbridge/natsbridge.dart';
final env = await NATSBridge.receive(
msg,
fileserverDownloadHandler: fetchWithBackoff,
maxRetries: 5,
baseDelay: 100,
maxDelay: 5000,
);
// Returns: Future<Map<String, dynamic>> with "payloads" key
```
#### Dart Web
```dart
import 'package:natsbridge/natsbridge.dart';
final env = await NATSBridge.receive(
msg,
fileserverDownloadHandler: fetchWithBackoff,
maxRetries: 5,
baseDelay: 100,
maxDelay: 5000,
);
// Returns: Future<Map<String, dynamic>> with "payloads" key
```
---
## Payload Types
| Type | Julia | JavaScript | Python | MicroPython | Description |
|------|-------|------------|--------|-------------|-------------|
| `text` | `String` | `string` | `str` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `dict` | JSON-serializable dictionaries |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `DataFrame`, `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | ⚠️ | Tabular data (JSON) - **Only table type in Browser** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `bytearray` | Image data (PNG, JPG) |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `bytearray` | Audio data (WAV, MP3) |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `bytearray` | Video data (MP4, AVI) |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `bytearray` | Generic binary data |
| Type | Julia | JavaScript | Python | Dart | Dart Web | MicroPython | Description |
|------|-------|------------|--------|------|----------|-------------|-------------|
| `text` | `String` | `string` | `str` | `String` | `String` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `Map` | `Map` | `dict` | JSON-serializable dictionaries |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | `List<Map>` (Desktop/Flutter) | ❌ (Browser incompatible) | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `DataFrame`, `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | `List<Map>` | `List<Map>` | ⚠️ | Tabular data (JSON) - **Only table type in Browser/Dart Web** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Uint8List` | `bytearray` | Image data (PNG, JPG) |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Uint8List` | `bytearray` | Audio data (WAV, MP3) |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Uint8List` | `bytearray` | Video data (MP4, AVI) |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `Uint8List` | `Uint8List` | `bytearray` | Generic binary data |
---
@@ -721,6 +833,7 @@ env, env_json_str = await NATSBridge.smartsend(
| **Julia** | `test/test_julia_*_sender.jl` | `test/test_julia_*_receiver.jl` |
| **JavaScript** | `test/test_js_*_sender.js` | `test/test_js_*_receiver.js` |
| **Python** | `test/test_py_*_sender.py` | `test/test_py_*_receiver.py` |
| **Dart** | `test/test_dart_*_sender.dart` | `test/test_dart_*_receiver.dart` |
### Run Tests
@@ -788,6 +901,30 @@ python3 test/test_py_table_sender.py
python3 test/test_py_table_receiver.py
```
#### Dart
```bash
# Text message exchange
dart test/test_dart_text_sender.dart
dart test/test_dart_text_receiver.dart
# Dictionary exchange
dart test/test_dart_dictionary_sender.dart
dart test/test_dart_dictionary_receiver.dart
# Binary transfer
dart test/test_dart_binary_sender.dart
dart test/test_dart_binary_receiver.dart
# Mixed payload types
dart test/test_dart_mix_payloads_sender.dart
dart test/test_dart_mix_payloads_receiver.dart
# Table exchange
dart test/test_dart_table_sender.dart
dart test/test_dart_table_receiver.dart
```
---
## Browser Deployment

View File

@@ -1,7 +1,7 @@
# Architecture Documentation: NATSBridge
**Version**: 1.4.0
**Date**: 2026-05-14
**Version**: 1.1.0
**Date**: 2026-03-23
**Status**: Active
**Ground Truth**: [`src/NATSBridge.jl`](../src/NATSBridge.jl)
**Architecture Level**: C4 Container Level
@@ -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**, **Rust**, 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**, 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,7 +57,6 @@ 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
@@ -65,14 +64,12 @@ 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
@@ -81,7 +78,6 @@ 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
```
@@ -89,20 +85,28 @@ flowchart TD
```mermaid
flowchart TD
subgraph "Client Container"
subgraph "Client Container"
Julia_Module[Julia NATSBridge Module]
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
@@ -111,7 +115,6 @@ 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
@@ -120,7 +123,6 @@ 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
@@ -137,31 +139,36 @@ flowchart TD
Serialize[_serialize_data]
Deserialize[_deserialize_data]
EnvelopeToJson[envelope_to_json]
BuildEnvelope[build_envelope]
BuildPayload[build_payload]
PublishMessage[publish_message]
FileServerUpload[fileserver_upload_handler]
FileServerDownload[fileserver_download_handler]
LogTrace[log_trace]
end
subgraph "Data Models"
Payload[msg_payload_v1 Struct]
Envelope[msg_envelope_v1 Struct]
Payload[MsgPayloadV1 Struct]
Envelope[MsgEnvelopeV1 Struct]
end
SmartSend --> Serialize
SmartSend --> EnvelopeToJson
SmartSend --> BuildEnvelope
SmartSend --> BuildPayload
SmartSend --> PublishMessage
SmartSend --> FileServerUpload
SmartReceive --> Deserialize
SmartReceive --> FileServerDownload
EnvelopeToJson --> Envelope
Serialize --> Payload
BuildEnvelope --> Envelope
BuildPayload --> Payload
style SmartSend fill:#d1fae5,stroke:#10b981
style SmartReceive fill:#d1fae5,stroke:#10b981
style PublishMessage fill:#fef3c7,stroke:#f59e0b
style FileServerUpload fill:#fef3c7,stroke:#f59e0b
style FileServerDownload fill:#fef3c7,stroke:#f59e0b
```
@@ -174,15 +181,15 @@ flowchart TD
| Component | Purpose | Platform Support |
|-----------|---------|------------------|
| **smartsend** | Send data via NATS with automatic transport selection, returns (envelope, json_string) for caller to publish | All |
| **smartreceive** | Receive and process NATS messages from JSON string | All |
| **smartsend** | Send data via NATS with automatic transport selection | All |
| **smartreceive** | Receive and process NATS messages | All |
| **_serialize_data** | Serialize data according to payload type | All |
| **_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/Rust) |
| **fileserver_download_handler** | Download payloads from HTTP server with exponential backoff | Desktop (Julia/JS/Python/Dart/Rust) |
| **plik_upload_file** | Upload a local file to Plik server from disk | Rust |
| **_build_envelope** | Build message envelope from payloads | All |
| **_build_payload** | Build payload object from serialized data | All |
| **publish_message** | Publish message to NATS subject | All |
| **fileserver_upload_handler** | Upload large payloads to HTTP server | Desktop (Julia/JS/Python/Dart) |
| **fileserver_download_handler** | Download payloads from HTTP server | Desktop (Julia/JS/Python/Dart) |
### Data Flow
@@ -204,7 +211,7 @@ flowchart TD
H --> L[Build envelope]
L --> M[Convert to JSON]
M --> N[Return envelope + JSON to caller]
M --> N[Publish to NATS]
style A fill:#f9f9f9,stroke:#333
style N fill:#e0e7ff,stroke:#3b82f6
@@ -297,7 +304,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/Rust) |
| `arrowtable` | Apache Arrow IPC | Arrow IPC stream | Base64/arrow-ipc | Desktop (Julia/Python/Node.js/Dart) |
| `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 |
@@ -430,9 +437,11 @@ end
JavaScript uses async/await for non-blocking I/O:
- **Class-based NATS Client**: Connection management with `keepAlive` support
- **Module-level Utilities**: Serialization functions
- **Native ArrayBuffer**: Binary data handling (Browser) / Buffer (Node.js)
- **Fetch API**: HTTP file server communication
- **Connection Pooling**: `NATSConnectionPool` for high-throughput scenarios
#### Node.js Implementation (natsbridge_ssr.js)
@@ -440,6 +449,36 @@ JavaScript uses async/await for non-blocking I/O:
- **Apache Arrow IPC**: Full support via `apache-arrow`
- **Buffer for binary data**: Native Node.js Buffer handling
```javascript
// Class-based NATS client with keepAlive support
class NATSClient {
constructor(url, keepAlive = false) {
this.url = url;
this.connection = null;
this.keepAlive = keepAlive;
}
async connect() {
if (this.connection) return this.connection;
this.connection = await nats.connect({ servers: this.url });
return this.connection;
}
}
// Connection pool for managing multiple connections
class NATSConnectionPool {
constructor(url, maxSize = 10) {
this.url = url;
this.maxSize = maxSize;
this.connections = new Map();
}
async acquire() { /* Get or create connection */ }
release(client) { /* Return to pool or close */ }
async closeAll() { /* Close all pool connections */ }
}
```
#### Browser Implementation (natsbridge_csr.js)
- **WebSocket NATS connections**: Uses `ws://` or `wss://` URLs via `nats.ws`
@@ -447,6 +486,23 @@ JavaScript uses async/await for non-blocking I/O:
- **Uint8Array for binary data**: Browser-compatible binary handling
- **Web Crypto API**: UUID generation via `crypto.getRandomValues()`
```javascript
// Class-based NATS client with keepAlive support
class NATSClient {
constructor(url, keepAlive = false) {
this.url = url; // ws:// or wss://
this.connection = null;
this.keepAlive = keepAlive;
}
async connect() {
if (this.connection) return this.connection;
this.connection = await nats.connect({ servers: this.url });
return this.connection;
}
}
```
### Python Architecture
Python uses classes for stateful operations:
@@ -534,63 +590,6 @@ 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<Arc<dyn FileUploadHandler>>,
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 = smartreceive(&String::from_utf8_lossy(&msg.payload), &Default::default()).await?;
// Access deserialized payloads by type
for payload in &envelope.payloads {
match payload.payload_type.as_str() {
"arrowtable" => { /* payload.data is base64-encoded Arrow IPC */ },
"text" => { /* payload.data is decoded text string */ },
"binary" | "image" | "audio" | "video" => { /* payload.data is base64-encoded binary */ },
_ => { /* other types */ }
}
}
}
```
---
## Scaling Architecture
@@ -741,7 +740,7 @@ for msg in sub.messages() {
|----------|---------|-------------|
| `NATS_URL` | `nats://localhost:4222` | NATS server URL |
| `FILESERVER_URL` | `http://localhost:8080` | HTTP file server URL |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
| `SIZE_THRESHOLD` | `1000000` | Size threshold in bytes |
### Container Deployment
@@ -827,7 +826,7 @@ flowchart TD
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, MicroPython 1.19+ |
---
@@ -835,23 +834,6 @@ flowchart TD
| Date | Version | Changes |
|------|---------|---------|
| 2026-05-14 | 1.4.0 | Updated Rust API to reflect `smartreceive` deserialization changes | All sections |
| - | - | `smartreceive` now stores deserialized data in `MsgPayloadV1.data` | specification.md:8 |
| - | - | Added `plik_upload_file` convenience function to component table | specification.md:13 |
| - | - | Fixed Rust payload access pattern (data is String, not Payload enum) | All sections |
| - | - | Fixed `SmartsendOptions.fileserver_upload_handler` type to `Arc<dyn FileUploadHandler>` | specification.md:13 |
| - | - | Removed `metadata` from link transport examples (now `None`/omitted) | specification.md:3 |
| - | - | Removed duplicate footer text | All sections |
| 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) |
| - | - | Updated component diagram to match actual module structure |
| - | - | Updated data flow to show smartsend returns JSON for caller to publish |
| - | - | Fixed SIZE_THRESHOLD default to 500,000 bytes |
| 2026-03-15 | 1.1.0 | JavaScript connection management |
| - | - | Added NATSClient with keepAlive support |
| - | - | Added NATSConnectionPool for connection reuse |
@@ -883,7 +865,6 @@ 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, file upload helpers | 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
@@ -906,13 +887,6 @@ 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 |
---
@@ -939,3 +913,7 @@ flowchart TD
---
*This architecture document is versioned and maintained in git alongside the codebase. All implementations must adhere to this architecture.*
---
*This architecture document is versioned and maintained in git alongside the codebase. All implementations must adhere to this architecture.*

View File

@@ -1,7 +1,7 @@
# Requirements Document: NATSBridge
**Version**: 1.2.0
**Date**: 2026-05-13
**Version**: 1.0.0
**Date**: 2026-03-23
**Status**: Active
**Ground Truth**: [`src/NATSBridge.jl`](../src/NATSBridge.jl)
@@ -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**, **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.
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.
### 1.2 User Stories (with acceptance criteria)
@@ -25,7 +25,6 @@ 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 |
@@ -52,7 +51,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, Rust, and MicroPython |
| Cross-platform interoperability | Seamless data exchange between Julia, JavaScript, Python, Dart, 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 |
@@ -89,11 +88,6 @@ 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
@@ -104,7 +98,6 @@ 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 |
---
@@ -125,8 +118,8 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| **FR-010** | Exponential backoff retry | System shall implement exponential backoff with configurable retries (default: 5, base_delay: 100ms, max_delay: 5000ms) for file server download failures |
| **FR-011** | Correlation ID propagation | System shall propagate correlation IDs through all message processing steps |
| **FR-012** | Message serialization | System shall serialize data types using Base64, JSON, or Arrow IPC encoding |
| **FR-013** | NATS publishing | System shall return JSON string representation for caller to publish to NATS subjects (caller is responsible for actual NATS publish) |
| **FR-014** | NATS subscription | System shall receive and process NATS messages by accepting JSON string from NATS payload |
| **FR-013** | NATS publishing | System shall publish messages to NATS subjects |
| **FR-014** | NATS subscription | System shall receive and process NATS messages |
---
@@ -196,14 +189,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` | `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 |
| `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 |
### 6.2 Encoding Requirements
@@ -227,7 +220,6 @@ 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
@@ -238,7 +230,6 @@ 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 |
---
@@ -333,11 +324,11 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
```julia
function smartsend(
subject::String,
data::AbstractArray{Tuple{String, T1, String}, 1};
broker_url::String = DEFAULT_BROKER_URL,
fileserver_url::String = DEFAULT_FILESERVER_URL,
data::AbstractArray{Tuple{String, Any, String}};
broker_url::String = "nats://localhost:4222",
fileserver_url::String = "http://localhost:8080",
fileserver_upload_handler::Function = plik_oneshot_upload,
size_threshold::Int = DEFAULT_SIZE_THRESHOLD,
size_threshold::Int = 1_000_000,
correlation_id::String = string(uuid4()),
msg_purpose::String = "chat",
sender_name::String = "NATSBridge",
@@ -345,18 +336,18 @@ function smartsend(
receiver_id::String = "",
reply_to::String = "",
reply_to_msg_id::String = "",
is_publish::Bool = true,
NATS_connection::Union{NATS.Connection, Nothing} = nothing,
msg_id::String = string(uuid4()),
sender_id::String = string(uuid4())
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
)::Tuple{msg_envelope_v1, String}
```
**Note**: NATS publishing is the caller's responsibility. `smartsend` returns `(env::msg_envelope_v1, env_json_str::String)`.
### 11.2 smartreceive Signature
```julia
function smartreceive(
msg_json_str::String;
msg::NATS.Msg;
fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5,
base_delay::Int = 100,
@@ -364,8 +355,6 @@ function smartreceive(
)::JSON.Object{String, Any}
```
**Note**: Pass `String(nats_msg.payload)` from NATS subscription to `smartreceive`.
---
## 12. Deployment Requirements
@@ -385,7 +374,7 @@ function smartreceive(
|----------|---------|-------------|
| `NATS_URL` | `nats://localhost:4222` | NATS server URL |
| `FILESERVER_URL` | `http://localhost:8080` | HTTP file server URL |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
| `SIZE_THRESHOLD` | `1000000` | Size threshold in bytes |
---
@@ -401,7 +390,7 @@ function smartreceive(
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, Browser (latest), MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Browser (latest), MicroPython 1.19+ |
---
@@ -409,13 +398,6 @@ function smartreceive(
| Date | Version | Changes |
|------|---------|---------|
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) |
| - | - | Fixed smartsend signature: removed is_publish, NATS_connection; added sender_name |
| - | - | Fixed smartreceive signature: takes msg_json_str::String instead of msg::NATS.Msg |
| - | - | Fixed size_threshold default from 1,000,000 to 500,000 |
| - | - | Updated FR-013/FR-014 to reflect caller responsibility for NATS publishing |
| - | - | Updated FR-008/FR-009 to include file path upload overload |
| - | - | Updated SIZE_THRESHOLD env var default to 500000 |
| 2026-03-23 | 1.0.0 | Updated to ASG Framework requirements structure |
---
@@ -428,7 +410,6 @@ 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

View File

@@ -1,7 +1,7 @@
# Specification: NATSBridge
**Version**: 1.2.0
**Date**: 2026-05-13
**Version**: 1.1.0
**Date**: 2026-03-23
**Status**: Active
**Ground Truth**: [`src/NATSBridge.jl`](../src/NATSBridge.jl)
**Specification Format**: JSON Schema + AsyncAPI
@@ -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**, **Rust**, 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**, 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()`
@@ -418,11 +418,11 @@ When `transport = "link"`, the `data` field contains a URL pointing to the uploa
```julia
function smartsend(
subject::String,
data::AbstractArray{Tuple{String, T1, String}, 1};
broker_url::String = DEFAULT_BROKER_URL,
fileserver_url::String = DEFAULT_FILESERVER_URL,
data::AbstractArray{Tuple{String, Any, String}};
broker_url::String = "nats://localhost:4222",
fileserver_url::String = "http://localhost:8080",
fileserver_upload_handler::Function = plik_oneshot_upload,
size_threshold::Int = DEFAULT_SIZE_THRESHOLD,
size_threshold::Int = 500_000,
correlation_id::String = string(uuid4()),
msg_purpose::String = "chat",
sender_name::String = "NATSBridge",
@@ -430,13 +430,13 @@ function smartsend(
receiver_id::String = "",
reply_to::String = "",
reply_to_msg_id::String = "",
is_publish::Bool = true,
NATS_connection::Union{NATS.Connection, Nothing} = nothing,
msg_id::String = string(uuid4()),
sender_id::String = string(uuid4())
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
)::Tuple{msg_envelope_v1, String}
```
**Note**: NATS publishing is the caller's responsibility. Returns `(env::msg_envelope_v1, env_json_str::String)`.
#### Python
```python
@@ -454,13 +454,13 @@ async def smartsend(
receiver_id: str = "",
reply_to: str = "",
reply_to_msg_id: str = "",
is_publish: bool = True,
nats_connection: Any = None,
msg_id: str = None,
sender_id: str = None
) -> Tuple[Dict, str]:
```
**Note**: NATS publishing is the caller's responsibility.
#### JavaScript (Node.js)
```typescript
@@ -479,14 +479,14 @@ async function smartsend(
receiver_id?: string;
reply_to?: string;
reply_to_msg_id?: string;
is_publish?: boolean;
nats_connection?: NATS.Connection;
msg_id?: string;
sender_id?: string;
}
): Promise<[Object, string]>;
```
**Note**: NATS publishing is the caller's responsibility.
#### JavaScript (Browser)
```typescript
@@ -505,13 +505,40 @@ async function smartsend(
receiver_id?: string;
reply_to?: string;
reply_to_msg_id?: string;
is_publish?: boolean;
nats_connection?: NATSClient | NATS.Connection;
msg_id?: string;
sender_id?: string;
}
): Promise<[Object, string]>;
```
**Note**: NATS publishing is the caller's responsibility.
// NATSClient class for connection management
class NATSClient {
constructor(url: string, keepAlive?: boolean);
connect(): Promise<NATS.Connection>;
publish(subject: string, message: string, correlationId: string): Promise<void>;
close(): Promise<void>;
getConnection(): NATS.Connection | null;
isConnected(): boolean;
}
// NATSConnectionPool for managing multiple connections
class NATSConnectionPool {
constructor(url: string, maxSize?: number);
acquire(): Promise<NATSClient>;
release(client: NATSClient): void;
closeAll(): Promise<void>;
}
// publishMessage function for manual publishing
async function publishMessage(
brokerUrlOrClient: string | NATSClient | NATS.Connection,
subject: string,
message: string,
correlationId: string,
closeConnection?: boolean
): Promise<void>;
```
#### MicroPython
@@ -519,13 +546,10 @@ async function smartsend(
def smartsend(
subject: str,
data: List[Tuple[str, Any, str]],
size_threshold: int = 100_000, # Lower threshold for memory constraints
**kwargs
) -> Tuple[Dict, str]:
```
**Note**: NATS publishing is the caller's responsibility.
#### Dart (Desktop/Flutter)
```dart
@@ -543,11 +567,12 @@ Future<[Map<String, dynamic>, String]> smartsend(
String receiverId = '',
String replyTo = '',
String replyToMsgId = '',
bool isPublish = true,
dynamic natsConnection,
String? msgId,
String? senderId,
}) async {
// Returns [envelope, jsonString]
// NATS publishing is caller's responsibility
}
```
@@ -568,82 +593,22 @@ Future<[Map<String, dynamic>, String]> smartsend(
String receiverId = '',
String replyTo = '',
String replyToMsgId = '',
bool isPublish = true,
dynamic natsConnection,
String? msgId,
String? senderId,
}) async {
// Returns [envelope, jsonString]
// NATS publishing is caller's responsibility
}
```
#### 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
```julia
function smartreceive(
msg_json_str::String; # Pass String(nats_msg.payload) from NATS subscription
msg::NATS.Msg;
fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5,
base_delay::Int = 100,
@@ -651,13 +616,11 @@ function smartreceive(
)::JSON.Object{String, Any}
```
**Note**: Input is JSON string from NATS message payload, not NATS.Msg directly.
#### Python
```python
async def smartreceive(
msg_json_str: str, # JSON string from NATS message payload
msg: Any,
fileserver_download_handler: Callable = fetch_with_backoff,
max_retries: int = 5,
base_delay: int = 100,
@@ -665,13 +628,11 @@ async def smartreceive(
) -> Dict[str, Any]:
```
**Note**: Input is JSON string from NATS message payload.
#### JavaScript (Node.js)
```typescript
async function smartreceive(
msg_json_str: string, // JSON string from NATS message payload
msg: Object,
options?: {
fileserver_download_handler?: Function;
max_retries?: number;
@@ -685,7 +646,7 @@ async function smartreceive(
```typescript
async function smartreceive(
msg_json_str: string, // JSON string from NATS message payload
msg: Object,
options?: {
fileserver_download_handler?: Function;
max_retries?: number;
@@ -695,22 +656,17 @@ async function smartreceive(
): Promise<Object>;
```
**Note**: Input is JSON string from NATS message payload.
#### MicroPython
```python
def smartreceive(msg_json_str: str, **kwargs) -> Dict[str, Any]:
def smartreceive(msg: Any, **kwargs) -> Dict[str, Any]:
```
**Note**: Input is JSON string from NATS message payload.
#### Dart (Desktop/Flutter)
```dart
Future<Map<String, dynamic>> smartreceive(
Map<String, dynamic> msg_json_str, // JSON object from NATS message payload
{
Map<String, dynamic> msg, {
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
@@ -724,8 +680,7 @@ Future<Map<String, dynamic>> smartreceive(
```dart
Future<Map<String, dynamic>> smartreceive(
Map<String, dynamic> msg_json_str, // JSON object from NATS message payload
{
Map<String, dynamic> msg, {
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
@@ -735,25 +690,6 @@ 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
@@ -767,12 +703,6 @@ function fileserver_upload_handler(
dataname::String,
data::Vector{UInt8}
)::Dict{String, Any}
# Overload: Upload file from disk
function fileserver_upload_handler(
file_server_url::String,
filepath::String
)::Dict{String, Any}
```
**Return Format**:
@@ -866,18 +796,6 @@ 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 |
@@ -900,7 +818,6 @@ 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
@@ -915,16 +832,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 | Rust | MicroPython |
|--------------|-------|---------|---------|--------|------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| Payload Type | Julia | Node.js | Browser | Python | Dart | MicroPython |
|--------------|-------|---------|---------|--------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
---
@@ -1062,13 +979,6 @@ 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
@@ -1121,8 +1031,6 @@ 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
@@ -1141,17 +1049,6 @@ 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 |
---
@@ -1160,12 +1057,6 @@ flowchart TD
| Date | Version | Changes | Requirement ID(s) |
|------|---------|---------|-------------------|
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) | All |
| - | - | Updated smartsend signatures: removed is_publish, nats_connection; added sender_name | FR-001 through FR-014 |
| - | - | Updated smartreceive signatures: takes msg_json_str::String instead of msg | FR-001 through FR-014 |
| - | - | Removed publishMessage function and NATSClient/NATSConnectionPool classes from browser section | FR-013, FR-014 |
| - | - | Added plik_oneshot_upload(filepath) overload to file server interface | FR-008, FR-009 |
| - | - | Fixed SIZE_THRESHOLD default to 500,000 bytes | FR-003, FR-004 |
| 2026-03-23 | 1.1.0 | Updated to ASG Framework specification guidelines | All |
| 2026-03-15 | 1.1.0 | Browser connection management | FR-001 through FR-014 |
| 2026-03-13 | 1.0.0 | Initial specification | FR-001 through FR-014, NFR-101 through NFR-405 |

View File

@@ -1,7 +1,7 @@
# Walkthrough: NATSBridge
**Version**: 1.4.0
**Date**: 2026-05-14
**Version**: 1.0.0
**Date**: 2026-03-23
**Status**: Active
**Ground Truth**: [`src/NATSBridge.jl`](../src/NATSBridge.jl)
@@ -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**, **Rust**, 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**, 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
@@ -213,25 +213,23 @@ NATSBridge builds the message envelope:
- **reply_to**: Tells backend where to send response
- **payloads array**: Contains all data with metadata for proper handling
#### Step 5: Publish to NATS (Caller's Responsibility)
#### Step 5: Publish to NATS
```javascript
// NATS publishing is the caller's responsibility
const conn = await NATS.connect({ servers: "ws://localhost:4222" });
await conn.publish("/agent/wine/api/v1/prompt", msgJson);
await NATSBridge.NATSClient.connect("ws://localhost:4222");
await NATSBridge.NATSClient.publish("/agent/wine/api/v1/prompt", msgJson);
```
**Rationale**:
- NATS provides low-latency message delivery
- JSON format ensures cross-platform compatibility
- `smartsend()` returns `(env, msgJson)` - caller handles publishing
#### Step 6: Julia Backend Receives Message
```julia
# Julia backend
nats_msg = NATS.subscription.next() # Get message from NATS
env = smartreceive(String(nats_msg.payload))
msg = NATS.subscription.next() # Get message from NATS
env = smartreceive(msg)
# env["payloads"] is now:
# [
@@ -341,7 +339,8 @@ const response = await plikOneshotUpload(
"transport": "link",
"encoding": "none",
"size": 10000000,
"data": "http://localhost:8080/file/UPLOAD_ID/FILE_ID/file"
"data": "http://localhost:8080/file/UPLOAD_ID/FILE_ID/file",
"metadata": {}
}
]
}
@@ -356,8 +355,8 @@ const response = await plikOneshotUpload(
```julia
# Julia backend
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
msg = NATS.subscription.next()
env = smartreceive(msg)
# NATSBridge automatically:
# 1. Extracts URL from payload
@@ -429,8 +428,8 @@ arrow_bytes = buf.getvalue()
```julia
# Julia backend
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
msg = NATS.subscription.next()
env = smartreceive(msg)
# env["payloads"][1] is now:
# ("data", DataFrame with id, name, score columns, "arrowtable")
@@ -462,159 +461,7 @@ env, msg_json = smartsend(
---
## 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};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
#[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 = smartreceive(
&String::from_utf8_lossy(&msg.payload),
&Default::default(),
).await.unwrap();
// Access deserialized payloads by type
for payload in &envelope.payloads {
match payload.payload_type.as_str() {
"arrowtable" => {
// Data is base64-encoded Arrow IPC bytes after smartreceive()
let arrow_bytes = BASE64.decode(&payload.data).unwrap();
println!("Received arrowtable payload ({} bytes)", arrow_bytes.len());
},
"text" => {
// Data is the decoded text string
println!("Message: {}", payload.data);
},
"image" | "audio" | "video" | "binary" => {
// Data is base64-encoded binary content
let bytes = BASE64.decode(&payload.data).unwrap();
println!("Received {} bytes of {} data", bytes.len(), payload.payload_type);
},
"dictionary" | "jsontable" => {
// Data is a JSON string
println!("Data: {}", payload.data);
},
_ => println!("Unknown payload type: {}", payload.payload_type),
}
}
}
}
```
**Rationale**:
- **serde serialization**: Automatic JSON deserialization to `MsgEnvelopeV1`
- **tokio runtime**: Efficient async I/O for NATS and HTTP operations
- **smartreceive deserialization**: Payload data is deserialized and stored as strings in `payload.data`
- **Type dispatch**: `payload_type` field determines how to interpret the `data` string
#### 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
## User Scenario 4: MicroPython Device
### Scenario Description
@@ -665,8 +512,8 @@ payload_b64 = base64.b64encode(json_bytes).decode('ascii')
```python
# Python backend
nats_msg = await nats_consumer.next()
env = await smartreceive(str(nats_msg.payload))
msg = await nats_consumer.next()
env = await smartreceive(msg)
# env["payloads"][0] is now:
# ("data", {"temperature": 25.5, "humidity": 60.0, ...}, "dictionary")
@@ -679,7 +526,7 @@ env = await smartreceive(str(nats_msg.payload))
---
## User Scenario 6: Cross-Platform Chat with Mixed Payloads
## User Scenario 5: Cross-Platform Chat with Mixed Payloads
### Scenario Description
@@ -714,8 +561,8 @@ const [env, msgJson] = await NATSBridge.smartsend(
```python
# Python (Backend)
nats_msg = await nats_consumer.next()
env = await smartreceive(str(nats_msg.payload))
msg = await nats_consumer.next()
env = await smartreceive(msg)
# env["payloads"] is now:
# [
@@ -733,8 +580,8 @@ env = await smartreceive(str(nats_msg.payload))
```julia
# Julia (Backend)
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
msg = NATS.subscription.next()
env = smartreceive(msg)
# env["payloads"] is now:
# [
@@ -879,7 +726,7 @@ log_trace(correlation_id, "Published to NATS")
|----------|---------|-------------|
| `NATS_URL` | `nats://localhost:4222` | NATS server URL |
| `FILESERVER_URL` | `http://localhost:8080` | HTTP file server URL |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
| `SIZE_THRESHOLD` | `1000000` | Size threshold in bytes |
---
@@ -914,7 +761,6 @@ 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, file upload helpers | 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) |
---
@@ -923,18 +769,6 @@ log_trace(correlation_id, "Published to NATS")
| Date | Version | Changes | Specification Reference |
|------|---------|---------|------------------------|
| 2026-05-14 | 1.4.0 | Updated Rust API to reflect `smartreceive` deserialization changes | All sections |
| - | - | `smartreceive` now stores deserialized data in `MsgPayloadV1.data` | specification.md:8 |
| - | - | Added `plik_upload_file` convenience function documentation | specification.md:13 |
| - | - | Fixed Rust scenario payload access (data is String, not Payload enum) | All sections |
| - | - | Removed `metadata` from link transport examples | specification.md:3 |
| 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 |
| - | - | Removed is_publish and nats_connection parameter references | All sections |
| 2026-03-23 | 1.0.0 | Updated to ASG Framework walkthrough guidelines | All sections |
| 2026-03-13 | 1.0.0 | Initial walkthrough documentation | specification.md:2-19 (all sections) |

56
etc.jl Normal file
View File

@@ -0,0 +1,56 @@
using HTTP, Arrow, JSON, DataFrames
df = DataFrame(id = 1:10_000, val = rand(10_000))
file_server_url = "http://192.168.88.104:8080"
url_getUploadID = "$file_server_url/api/upload"
function upload_to_plik(url, df)
# 1. Build the Request object manually
headers = [
"X-Plik-TTL" => "5m",
"Content-Type" => "application/octet-stream",
"Transfer-Encoding" => "chunked"
]
# We create a request with an empty body, but we'll stream into it
req = HTTP.Request("POST", url, headers)
# 2. Open the connection manually to get a raw Stream
local_url = ""
HTTP.open("POST", url, headers) do stream
# WRITE PHASE
# Arrow.write handles the 'chunked' encoding automatically
Arrow.write(stream, df; file=false)
# CLOSE WRITE / START READ
# This is the critical hand-off.
# We tell the kernel we are done sending.
HTTP.closewrite(stream)
# Now we wait for the server's response
resp = HTTP.startread(stream)
# Handle the body
if resp.status == 200 || resp.status == 201
payload = read(stream, String)
# Depending on Plik version, it might return the URL directly
# or a JSON object. Adjust accordingly:
try
local_url = JSON.parse(payload)["url"]
catch
local_url = payload # Fallback if it's a raw string
end
else
error("Plik rejected upload with status: $(resp.status)")
end
end
return local_url
end
url = upload_to_plik(url_getUploadID, df)

View File

@@ -1,96 +0,0 @@
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);
}
}
}

View File

@@ -1,70 +0,0 @@
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);
}
}
}

View File

@@ -43,7 +43,7 @@
module NATSBridge
using JSON, Arrow, HTTP, UUIDs, Dates, Base64, PrettyPrinting, DataFrames
using NATS, JSON, Arrow, HTTP, UUIDs, Dates, Base64, PrettyPrinting, DataFrames
# ---------------------------------------------- 100 --------------------------------------------- #
# Constants
@@ -340,29 +340,25 @@ end
""" smartsend - Send data either directly via NATS or via a fileserver URL, depending on payload size
This function intelligently routes data delivery based on payload size relative to a threshold.
If the serialized payload is smaller than `size_threshold`, it encodes the data as Base64 and constructs a "direct" msg_payload_v1.
Otherwise, it uploads the data to a fileserver (by default using `plik_oneshot_upload`) and constructs a "link" msg_payload_v1 with the download URL.
If the serialized payload is smaller than `size_threshold`, it encodes the data as Base64 and publishes directly over NATS.
Otherwise, it uploads the data to a fileserver (by default using `plik_oneshot_upload`) and publishes only the download URL over NATS.
The function accepts a list of (dataname, data, type) tuples as input and processes each payload individually.
Each payload can have a different type, enabling mixed-content messages (e.g., chat with text, images, audio).
This function creates and returns the msg_envelope_v1 and its JSON string representation only.
NATS publishing must be performed by the caller.
# Function Workflow:
1. Iterates through the list of (dataname, data, type) tuples
2. For each payload: extracts the type from the tuple and serializes accordingly
3. Compares the serialized size against `size_threshold`
4. For small payloads: encodes as Base64, constructs a "direct" msg_payload_v1
5. For large payloads: uploads to the fileserver, constructs a "link" msg_payload_v1 with the URL
6. Constructs msg_envelope_v1 with all payloads and metadata
7. Converts envelope to JSON string and returns (NATS publishing is handled by the caller)
6. Converts envelope to JSON string and optionally publishes to NATS
# Arguments:
- `subject::String` - NATS subject to publish the message to
- `data::AbstractArray{Tuple{String, T1, String}, 1}` - List of (dataname, data, type) tuples to send
- `data::AbstractArray{Tuple{String, Any, String}}` - List of (dataname, data, type) tuples to send
- `dataname::String` - Name of the payload
- `data::T1` - The actual data to send (any type supported by `_serialize_data`)
- `data::Any` - The actual data to send
- `payload_type::String` - Payload type: "text", "dictionary", "arrowtable", "jsontable", "image", "audio", "video", "binary"
- No standalone `type` parameter - type is specified per payload
@@ -371,15 +367,17 @@ NATS publishing must be performed by the caller.
- `fileserver_url = DEFAULT_FILESERVER_URL` - URL of the HTTP file server for large payloads
- `fileserver_upload_handler::Function = plik_oneshot_upload` - Function to handle fileserver uploads (must return Dict with "status", "uploadid", "fileid", "url" keys)
- `size_threshold::Int = DEFAULT_SIZE_THRESHOLD` - Threshold in bytes separating direct vs link transport
- `correlation_id::String = string(uuid4())` - Correlation ID for tracing (auto-generated UUID)
- `msg_purpose::String = "chat"` - Purpose of the message: "ACK", "NACK", "updateStatus", "shutdown", "chat", etc.
- `sender_name::String = "NATSBridge"` - Name of the sender
- `receiver_name::String = ""` - Name of the receiver (empty string means broadcast)
- `correlation_id::String = string(uuid4())` - Correlation ID for tracing (auto-generated UUID)
- `msg_purpose::String = "chat"` - Purpose of the message: "ACK", "NACK", "updateStatus", "shutdown", "chat", etc.
- `sender_name::String = "NATSBridge"` - Name of the sender
- `receiver_name::String = ""` - Name of the receiver (empty string means broadcast)
- `receiver_id::String = ""` - UUID of the receiver (empty string means broadcast)
- `reply_to::String = ""` - Topic to reply to (empty string if no reply expected)
- `reply_to_msg_id::String = ""` - Message ID this message is replying to
- `msg_id::String = string(uuid4())` - Message ID (auto-generated UUID if not provided)
- `sender_id::String = string(uuid4())` - Sender ID (auto-generated UUID if not provided)
- `reply_to::String = ""` - Topic to reply to (empty string if no reply expected)
- `reply_to_msg_id::String = ""` - Message ID this message is replying to
- `is_publish::Bool = true` - Whether to automatically publish the message to NATS
- `NATS_connection::Union{NATS.Connection, Nothing} = nothing` - Pre-existing NATS connection (if provided, uses this connection instead of creating a new one; saves connection establishment overhead)
- `msg_id::String = string(uuid4())` - Message ID (auto-generated UUID if not provided)
- `sender_id::String = string(uuid4())` - Sender ID (auto-generated UUID if not provided)
# Return:
- `::Tuple{msg_envelope_v1, String}` - A tuple containing:
@@ -414,9 +412,8 @@ env, msg_json = smartsend("chat.subject", [
("audio_clip", audio_data, "audio")
])
# Publish the JSON string directly using NATS (manual publish)
# conn = NATS.connect(broker_url)
# NATS.publish(conn, subject, env_json_str)
# Publish the JSON string directly using NATS request-reply pattern
# reply = NATS.request(broker_url, subject, env_json_str; reply_to=reply_to_topic)
```
"""
function smartsend(
@@ -441,6 +438,8 @@ function smartsend(
receiver_id::String = "",
reply_to::String = "",
reply_to_msg_id::String = "",
is_publish::Bool = true, # some time the user want to get env and env_json_str from this function without publishing the msg
NATS_connection::Union{NATS.Connection, Nothing} = nothing, # a provided connection saves establishing connection overhead.
msg_id::String = string(uuid4()), # Message ID
sender_id::String = string(uuid4()) # Sender ID
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
@@ -540,15 +539,13 @@ function smartsend(
)
env_json_str = envelope_to_json(env) # Convert envelope to JSON
# if is_publish == false
# # skip publish a message
# elseif is_publish == true && NATS_connection === nothing
# # Publish message to NATS using new connection
# publish_message(broker_url, subject, env_json_str, correlation_id)
# elseif is_publish == true && NATS_connection !== nothing
# # Publish message to NATS using existing connection
# publish_message(NATS_connection, subject, env_json_str, correlation_id)
# end
if is_publish == false
# skip publish a message
elseif is_publish == true && NATS_connection === nothing
publish_message(broker_url, subject, env_json_str, correlation_id) # Publish message to NATS
elseif is_publish == true && NATS_connection !== nothing
publish_message(NATS_connection, subject, env_json_str, correlation_id) # Publish message to NATS
end
return (env, env_json_str)
end
@@ -703,73 +700,73 @@ function _serialize_data(data::Any, payload_type::String)
end
# """ publish_message - Publish message to NATS
# This function publishes a message to a NATS subject with proper
# connection management and logging.
""" publish_message - Publish message to NATS
This function publishes a message to a NATS subject with proper
connection management and logging.
# # Arguments:
# - `broker_url::String` - NATS server URL (e.g., "nats://localhost:4222")
# - `subject::String` - NATS subject to publish to (e.g., "/agent/wine/api/v1/prompt")
# - `message::String` - JSON message to publish
# - `correlation_id::String` - Correlation ID for tracing and logging
# Arguments:
- `broker_url::String` - NATS server URL (e.g., "nats://localhost:4222")
- `subject::String` - NATS subject to publish to (e.g., "/agent/wine/api/v1/prompt")
- `message::String` - JSON message to publish
- `correlation_id::String` - Correlation ID for tracing and logging
# # Return:
# - `nothing` - This function performs publishing but returns nothing
# Return:
- `nothing` - This function performs publishing but returns nothing
# # Example
# ```jldoctest
# using NATS
# Example
```jldoctest
using NATS
# # Prepare JSON message
# message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
# Prepare JSON message
message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
# # Publish to NATS
# publish_message("nats://localhost:4222", "my.subject", message, "abc123")
# ```
# """
# function publish_message(broker_url::String, subject::String, message::String, correlation_id::String)
# conn = NATS.connect(broker_url) # Create NATS connection
# publish_message(conn, subject, message, correlation_id)
# end
# Publish to NATS
publish_message("nats://localhost:4222", "my.subject", message, "abc123")
```
"""
function publish_message(broker_url::String, subject::String, message::String, correlation_id::String)
conn = NATS.connect(broker_url) # Create NATS connection
publish_message(conn, subject, message, correlation_id)
end
# """ publish_message - Publish message to NATS using pre-existing connection
# This function publishes a message to a NATS subject using a pre-existing NATS connection,
# avoiding the overhead of connection establishment.
""" publish_message - Publish message to NATS using pre-existing connection
This function publishes a message to a NATS subject using a pre-existing NATS connection,
avoiding the overhead of connection establishment.
# # Arguments:
# - `conn::NATS.Connection` - Pre-existing NATS connection
# - `subject::String` - NATS subject to publish to (e.g., "/agent/wine/api/v1/prompt")
# - `message::String` - JSON message to publish
# - `correlation_id::String` - Correlation ID for tracing and logging
# Arguments:
- `conn::NATS.Connection` - Pre-existing NATS connection
- `subject::String` - NATS subject to publish to (e.g., "/agent/wine/api/v1/prompt")
- `message::String` - JSON message to publish
- `correlation_id::String` - Correlation ID for tracing and logging
# # Return:
# - `nothing` - This function performs publishing but returns nothing
# Return:
- `nothing` - This function performs publishing but returns nothing
# # Example
# ```jldoctest
# using NATS
# Example
```jldoctest
using NATS
# # Prepare JSON message
# message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
# Prepare JSON message
message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
# # Create connection once and reuse for multiple publishes
# conn = NATS.connect("nats://localhost:4222")
# publish_message(conn, "my.subject", message, "abc123")
# # Connection is automatically drained after publish
# ```
# Create connection once and reuse for multiple publishes
conn = NATS.connect("nats://localhost:4222")
publish_message(conn, "my.subject", message, "abc123")
# Connection is automatically drained after publish
```
# # Use Case:
# Use this version when you already have an established NATS connection and want to publish
# multiple messages without the overhead of creating a new connection for each publish.
# """
# function publish_message(conn::NATS.Connection, subject::String, message::String, correlation_id::String)
# try
# NATS.publish(conn, subject, message) # Publish message to NATS
# log_trace(correlation_id, "Message published to $subject") # Log successful publish
# finally
# NATS.drain(conn) # Ensure connection is closed properly
# end
# end
# Use Case:
Use this version when you already have an established NATS connection and want to publish
multiple messages without the overhead of creating a new connection for each publish.
"""
function publish_message(conn::NATS.Connection, subject::String, message::String, correlation_id::String)
try
NATS.publish(conn, subject, message) # Publish message to NATS
log_trace(correlation_id, "Message published to $subject") # Log successful publish
finally
NATS.drain(conn) # Ensure connection is closed properly
end
end
""" smartreceive - Receive and process messages from NATS
@@ -786,7 +783,7 @@ A HTTP file server is required along with its download function.
5. For link transport: fetches data from URL with exponential backoff, then deserializes
# Arguments:
- `msg_json_str::String` - JSON string from NATS message payload (e.g., `String(nats_msg.payload)`)
- `msg::NATS.Msg` - NATS message to process
# Keyword Arguments:
- `fileserver_download_handler::Function = _fetch_with_backoff` - Function to handle downloading data from file server URLs
@@ -801,21 +798,19 @@ A HTTP file server is required along with its download function.
```jldoctest
# Receive and process message
msg = nats_message # NATS message
msg_json_str = String(msg.payload)
env = smartreceive(msg_json_str; fileserver_download_handler=_fetch_with_backoff, max_retries=5, base_delay=100, max_delay=5000)
env = smartreceive(msg; fileserver_download_handler=_fetch_with_backoff, max_retries=5, base_delay=100, max_delay=5000)
# env["payloads"] = [("dataname1", data1, "type1"), ("dataname2", data2, "type2"), ...]
```
"""
function smartreceive(
msg_json_str::String; # get it from String(nats_msg.payload)
msg::NATS.Msg;
fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5,
base_delay::Int = 100,
max_delay::Int = 5000
)::JSON.Object{String, Any}
# Parse the JSON envelope
env_json_obj = JSON.parse(msg_json_str)
env_json_obj = JSON.parse(String(msg.payload))
log_trace(env_json_obj["correlation_id"], "Processing received message") # Log message processing start
# Process all payloads in the envelope

782
src/natsbridge.dart Normal file
View File

@@ -0,0 +1,782 @@
/// NATSBridge - Cross-Platform Bi-Directional Data Bridge
/// Dart Implementation (Desktop/Flutter/Web)
///
/// This module provides functionality for sending and receiving data across network boundaries
/// using NATS as the message bus, with support for both direct payload transport and
/// URL-based transport for larger payloads.
///
/// Supported payload types: "text", "dictionary", "arrowtable", "jsontable", "image", "audio", "video", "binary"
///
/// Dart-specific features:
/// - Apache Arrow IPC support via dart-arrow (Desktop/Flutter only)
/// - TCP NATS connections via nats package (nats:// or tls:// URLs)
/// - WebSocket NATS support for Dart Web (ws:// or wss:// URLs)
/// - HTTP file server communication via http package
/// - Uint8List for binary data handling
///
/// Platform-specific notes:
/// - Desktop/Flutter: Full feature set including Arrow IPC
/// - Dart Web: JSON table only (no Arrow IPC), uses WebSocket NATS
///
/// @package natsbridge
import 'dart:async';
import 'dart:io';
import 'dart:typed_data';
import 'dart:util';
import 'dart:convert';
import 'package:http/http.dart' as http;
import 'package:uuid/uuid.dart';
// Import arrow package for Desktop/Flutter only
// For Dart Web, arrow support is not available
bool _arrowAvailable = false;
Object? _arrow;
Object? _ipc;
void _initArrow() {
try {
// Only available in Desktop/Flutter, not in Dart Web
// This will throw if dart-arrow is not available
// In a real implementation, you would use conditional imports
_arrowAvailable = false;
} catch (e) {
_arrowAvailable = false;
}
}
// ---------------------------------------------- UUID Helper ---------------------------------------------- //
/// Generate UUID v4
String _uuidv4() {
return const Uuid().v4();
}
// ---------------------------------------------- Constants ---------------------------------------------- //
/// Default size threshold for switching from direct to link transport (0.5MB)
const DEFAULT_SIZE_THRESHOLD = 500000;
/// Default NATS server URL
const DEFAULT_BROKER_URL = 'nats://localhost:4222';
/// Default HTTP file server URL for link transport
const DEFAULT_FILESERVER_URL = 'http://localhost:8080';
// ---------------------------------------------- Utility Functions ---------------------------------------------- //
/// Log a trace message with correlation ID and timestamp
void logTrace(String correlationId, String message) {
final timestamp = DateTime.now().toUtc().toIsoString();
print('[$timestamp] [Correlation: $correlationId] $message');
}
// ---------------------------------------------- Serialization Functions ---------------------------------------------- //
/// Serialize data according to specified format
Future<Uint8List> _serializeData(dynamic data, String payloadType) async {
if (payloadType == 'text') {
if (data is String) {
return Uint8List.fromList(utf8.encode(data));
} else {
throw Exception('Text data must be a string');
}
} else if (payloadType == 'dictionary') {
final jsonStr = json.encode(data);
return Uint8List.fromList(utf8.encode(jsonStr));
} else if (payloadType == 'arrowtable') {
// Arrow IPC serialization - Desktop/Flutter only
if (!_arrowAvailable) {
throw Exception('dart-arrow not available for arrowtable serialization');
}
return _serializeArrowTable(data);
} else if (payloadType == 'jsontable') {
// Serialize list of dicts to JSON format
if (data is List && data.every((row) => row is Map)) {
final jsonStr = json.encode(data);
return Uint8List.fromList(utf8.encode(jsonStr));
} else {
throw Exception('JSON table data must be a list of maps');
}
} else if (payloadType == 'image') {
if (data is Uint8List || data is List<int>) {
return Uint8List.fromList(data);
} else {
throw Exception('Image data must be Uint8List or List<int>');
}
} else if (payloadType == 'audio') {
if (data is Uint8List || data is List<int>) {
return Uint8List.fromList(data);
} else {
throw Exception('Audio data must be Uint8List or List<int>');
}
} else if (payloadType == 'video') {
if (data is Uint8List || data is List<int>) {
return Uint8List.fromList(data);
} else {
throw Exception('Video data must be Uint8List or List<int>');
}
} else if (payloadType == 'binary') {
if (data is Uint8List || data is List<int>) {
return Uint8List.fromList(data);
} else {
throw Exception('Binary data must be Uint8List or List<int>');
}
} else {
throw Exception('Unknown payload_type: $payloadType');
}
}
/// Helper function to serialize table data to Arrow IPC
Future<Uint8List> _serializeArrowTable(List<Map> data) async {
if (!_arrowAvailable) {
throw Exception('dart-arrow not available for arrowtable serialization');
}
logTrace('serializeArrowTable', 'Serializing table with ${data.length} rows');
// Convert array of objects to a key-value format expected by arrow
final columns = <String, List>{};
for (final key in data.isNotEmpty ? data[0].keys.toList() : []) {
columns[key] = data.map((row) => row[key]).toList();
}
logTrace('serializeArrowTable', 'Columns: ${columns.keys.join(', ')}');
// In a real implementation with dart-arrow, you would:
// 1. Create arrow fields from column types
// 2. Create arrow arrays from column data
// 3. Create an arrow table
// 4. Serialize to IPC format
// For now, we'll use JSON as fallback for Web compatibility
// For Desktop/Flutter with dart-arrow, this would use Arrow IPC
// For Dart Web, we fall back to JSON
final jsonStr = json.encode(data);
return Uint8List.fromList(utf8.encode(jsonStr));
}
/// Deserialize bytes to data based on type
Future<dynamic> _deserializeData(Uint8List data, String payloadType, String correlationId) async {
logTrace(correlationId, 'deserializeData: type=$payloadType, bufferLength=${data.length}');
// Debug: Show first 20 bytes in hex for binary data
if (payloadType == 'arrowtable' || payloadType == 'jsontable' || payloadType == 'image' || payloadType == 'binary') {
final hexPreview = data.length >= 20
? data.sublist(0, 20).map((b) => b.toRadixString(16).padLeft(2, '0')).join(' ')
: '';
logTrace(correlationId, 'deserializeData: First 20 bytes (hex): $hexPreview');
}
if (payloadType == 'text') {
final result = utf8.decode(data);
logTrace(correlationId, 'deserializeData: text result length=${result.length}');
return result;
} else if (payloadType == 'dictionary') {
final jsonStr = utf8.decode(data);
final result = json.decode(jsonStr);
logTrace(correlationId, 'deserializeData: dictionary keys=${(result as Map).keys.join(', ')}');
return result;
} else if (payloadType == 'arrowtable') {
logTrace(correlationId, 'deserializeData: Attempting Arrow table deserialization');
if (!_arrowAvailable) {
// Fallback to JSON for Web
final jsonStr = utf8.decode(data);
final result = json.decode(jsonStr);
return result;
}
// In a real implementation with dart-arrow, you would:
// 1. Read from IPC buffer
// 2. Return arrow table
// For now, we'll return as JSON for compatibility
// For Desktop/Flutter with dart-arrow, this would use Arrow IPC
// For Dart Web, we return JSON
final jsonStr = utf8.decode(data);
final result = json.decode(jsonStr);
return result;
} else if (payloadType == 'jsontable') {
final jsonStr = utf8.decode(data);
final result = json.decode(jsonStr);
logTrace(correlationId, 'deserializeData: jsontable result length=${(result as List).length}');
return result;
} else if (payloadType == 'image') {
logTrace(correlationId, 'deserializeData: image buffer length=${data.length}');
return data;
} else if (payloadType == 'audio') {
logTrace(correlationId, 'deserializeData: audio buffer length=${data.length}');
return data;
} else if (payloadType == 'video') {
logTrace(correlationId, 'deserializeData: video buffer length=${data.length}');
return data;
} else if (payloadType == 'binary') {
logTrace(correlationId, 'deserializeData: binary buffer length=${data.length}');
return data;
} else {
throw Exception('Unknown payload_type: $payloadType');
}
}
// ---------------------------------------------- File Server Handlers ---------------------------------------------- //
/// Upload data to plik server in one-shot mode
Future<Map<String, dynamic>> plikOneshotUpload(
String fileServerUrl,
String dataname,
Uint8List data,
) async {
final urlGetUploadID = '$fileServerUrl/upload';
// Get upload id
final response1 = await http.post(
Uri.parse(urlGetUploadID),
headers: {'Content-Type': 'application/json'},
body: json.encode({'OneShot': true}),
);
if (response1.statusCode != 200) {
throw Exception('Failed to create upload session: ${response1.statusCode}');
}
final responseJson1 = json.decode(response1.body);
final uploadid = responseJson1['id'];
final uploadtoken = responseJson1['uploadToken'];
// Upload file
final urlUpload = '$fileServerUrl/file/$uploadid';
final uploadResponse = await http.post(
Uri.parse(urlUpload),
headers: {'X-UploadToken': uploadtoken},
body: {
'file': http.MultipartFile.fromBytes(
'file',
data,
filename: dataname,
contentType: MediaType('application', 'octet-stream'),
),
},
);
if (uploadResponse.statusCode != 200) {
throw Exception('Failed to upload file: ${uploadResponse.statusCode}');
}
final uploadJson = json.decode(uploadResponse.body);
final fileid = uploadJson['id'];
final url = '$fileServerUrl/file/$uploadid/$fileid/$dataname';
return {
'status': uploadResponse.statusCode,
'uploadid': uploadid,
'fileid': fileid,
'url': url,
};
}
/// Fetch data from URL with exponential backoff
Future<Uint8List> fetchWithBackoff(
String url,
int maxRetries,
int baseDelay,
int maxDelay,
String correlationId,
) async {
var delay = baseDelay;
for (var attempt = 1; attempt <= maxRetries; attempt++) {
try {
final response = await http.get(Uri.parse(url));
if (response.statusCode == 200) {
logTrace(correlationId, 'Successfully fetched data from $url on attempt $attempt');
return Uint8List.fromList(response.bodyBytes);
} else {
throw Exception('Failed to fetch: ${response.statusCode}');
}
} catch (e) {
logTrace(correlationId, 'Attempt $attempt failed: ${e.runtimeType} - ${e.toString()}');
if (attempt < maxRetries) {
await Future.delayed(Duration(milliseconds: delay));
delay = (delay * 2).clamp(baseDelay, maxDelay);
}
}
}
throw Exception('Failed to fetch data after $maxRetries attempts');
}
// ---------------------------------------------- NATS Client ---------------------------------------------- //
/// NATS client wrapper for connection management
/// Supports both single-use and persistent connection modes
class NATSClient {
final String url;
Object? _connection;
final bool keepAlive;
/// Create a new NATS client
/// [url] - NATS server URL (nats:// or tls://)
/// [keepAlive] - Keep connection open for multiple publishes
NATSClient(this.url, {this.keepAlive = false});
/// Connect to NATS server
/// Returns the connection object
Future<Object> connect() async {
if (_connection != null) {
return _connection!;
}
try {
// Import nats package dynamically
final nats = await _loadNatsPackage();
_connection = await nats.connect(url);
return _connection!;
} catch (e) {
throw Exception('Failed to connect to NATS server: $e');
}
}
/// Publish message to NATS subject
Future<void> publish(String subject, String message, String correlationId) async {
if (_connection == null) {
await connect();
}
try {
final nats = await _loadNatsPackage();
await nats.publish(subject, message);
logTrace(correlationId, 'Message published to $subject');
} catch (e) {
throw Exception('Failed to publish message: $e');
}
}
/// Close the NATS connection
Future<void> close() async {
if (_connection != null) {
try {
final nats = await _loadNatsPackage();
await nats.close();
} catch (e) {
// Ignore errors on close
}
_connection = null;
}
}
/// Get the current connection
Object? getConnection() {
return _connection;
}
/// Check if connected
bool isConnected() {
return _connection != null;
}
/// Load the nats package dynamically
Future<dynamic> _loadNatsPackage() async {
// In a real implementation, you would use conditional imports
// For now, we'll throw an error indicating the package needs to be imported
// This is a limitation of Dart's dynamic import system
throw Exception('nats package not available. Please ensure dart-nats is installed.');
}
}
/// Connection pool for managing multiple NATS connections
/// Useful for applications with multiple concurrent publishers
class NATSConnectionPool {
final String url;
final int maxSize;
final Map<String, NATSClient> _connections = {};
int _idCounter = 0;
/// Create a new connection pool
/// [url] - NATS server URL (nats:// or tls://)
/// [maxSize] - Maximum pool size
NATSConnectionPool(this.url, {this.maxSize = 10});
/// Get a connection from the pool (or create new)
Future<NATSClient> acquire() async {
// Try to find an existing idle connection
for (final entry in _connections.entries) {
if (entry.value.isConnected()) {
return entry.value;
}
}
// Create new connection if under limit
if (_connections.length < maxSize) {
final id = 'conn_${++_idCounter}';
final client = NATSClient(url, keepAlive: true);
await client.connect();
_connections[id] = client;
return client;
}
// Pool exhausted - create new connection (caller should close when done)
final client = NATSClient(url, keepAlive: false);
await client.connect();
return client;
}
/// Return a connection to the pool
void release(NATSClient client) {
// Only return persistent connections
if (client.keepAlive && client.isConnected()) {
// Connection already in pool, do nothing
return;
}
// Non-persistent connection - close it
client.close();
}
/// Close all connections in the pool
Future<void> closeAll() async {
for (final entry in _connections.entries) {
await entry.value.close();
}
_connections.clear();
}
}
// ---------------------------------------------- Core Functions ---------------------------------------------- //
/// Build message envelope from payloads and metadata
Map<String, dynamic> _buildEnvelope(
String subject,
List<Map<String, dynamic>> payloads,
Map<String, dynamic> options,
) {
return {
'correlation_id': options['correlation_id'],
'msg_id': options['msg_id'],
'timestamp': DateTime.now().toUtc().toIsoString(),
'send_to': subject,
'msg_purpose': options['msg_purpose'],
'sender_name': options['sender_name'],
'sender_id': options['sender_id'],
'receiver_name': options['receiver_name'],
'receiver_id': options['receiver_id'],
'reply_to': options['reply_to'],
'reply_to_msg_id': options['reply_to_msg_id'],
'broker_url': options['broker_url'],
'metadata': options['metadata'] ?? {},
'payloads': payloads,
};
}
/// Build payload object from serialized data
Map<String, dynamic> _buildPayload(
String dataname,
String payloadType,
Uint8List payloadBytes,
String transport,
dynamic data,
) {
// Determine encoding based on payload type (matching Julia/JS implementation)
String encoding = 'base64';
if (payloadType == 'jsontable') {
encoding = 'json';
} else if (payloadType == 'arrowtable') {
encoding = 'arrow-ipc';
}
return {
'id': _uuidv4(),
'dataname': dataname,
'payload_type': payloadType,
'transport': transport,
'encoding': encoding,
'size': payloadBytes.length,
'data': data,
'metadata': transport == 'direct' ? {'payload_bytes': payloadBytes.length} : {},
};
}
/// Publish message to NATS
Future<void> publishMessage(
dynamic brokerUrlOrClient,
String subject,
String message,
String correlationId,
) async {
if (brokerUrlOrClient is NATSClient) {
final client = brokerUrlOrClient;
await client.publish(subject, message, correlationId);
await client.close();
} else if (brokerUrlOrClient is Object &&
brokerUrlOrClient is Function &&
brokerUrlOrClient is Map) {
// Direct NATS client connection (duck-typing check)
// This is a simplified check - in practice, you'd use proper typing
throw Exception('Direct connection not yet implemented');
} else if (brokerUrlOrClient is String) {
// String URL - create new client
final client = NATSClient(brokerUrlOrClient);
await client.connect();
await client.publish(subject, message, correlationId);
await client.close();
} else {
throw Exception('Invalid broker URL or client');
}
}
/// Send data via NATS with automatic transport selection
///
/// This function intelligently routes data delivery based on payload size.
/// If the serialized payload is smaller than size_threshold, it encodes the data as Base64
/// and publishes directly over NATS. Otherwise, it uploads the data to a fileserver
/// and publishes only the download URL over NATS.
///
/// [subject] - NATS subject to publish the message to
/// [data] - List of [dataname, data, type] lists to send
/// - dataname: Name of the payload
/// - data: The actual data to send
/// - type: Payload type: "text", "dictionary", "arrowtable", "jsontable", "image", "audio", "video", "binary"
/// [options] - Optional configuration
///
/// Returns a Future that completes with a tuple of [envelope, env_json_str]
Future<List<dynamic>> smartsend(
String subject,
List<List<dynamic>> data, {
String brokerUrl = DEFAULT_BROKER_URL,
String fileserverUrl = DEFAULT_FILESERVER_URL,
Function? fileserverUploadHandler,
int sizeThreshold = DEFAULT_SIZE_THRESHOLD,
String? correlationId,
String msgPurpose = 'chat',
String senderName = 'NATSBridge',
String receiverName = '',
String receiverId = '',
String replyTo = '',
String replyToMsgId = '',
bool isPublish = true,
dynamic natsConnection,
String? msgId,
String? senderId,
}) async {
final actualCorrelationId = correlationId ?? _uuidv4();
final actualMsgId = msgId ?? _uuidv4();
final actualSenderId = senderId ?? _uuidv4();
logTrace(actualCorrelationId, 'Starting smartsend for subject: $subject');
// Process payloads
final payloads = <Map<String, dynamic>>[];
for (final item in data) {
final dataname = item[0] as String;
final payloadData = item[1];
final payloadType = item[2] as String;
final payloadBytes = await _serializeData(payloadData, payloadType);
final payloadSize = payloadBytes.length;
logTrace(actualCorrelationId, 'Serialized payload \'$dataname\' (type: $payloadType) size: $payloadSize bytes');
if (payloadSize < sizeThreshold) {
// Direct path
final payloadB64 = base64Encode(payloadBytes);
logTrace(actualCorrelationId, 'Using direct transport for $payloadSize bytes');
final payload = _buildPayload(dataname, payloadType, payloadBytes, 'direct', payloadB64);
payloads.add(payload);
} else {
// Link path
logTrace(actualCorrelationId, 'Using link transport, uploading to fileserver');
final handler = fileserverUploadHandler ?? plikOneshotUpload;
final response = await handler(fileserverUrl, dataname, payloadBytes);
if (response['status'] != 200) {
throw Exception('Failed to upload data to fileserver: ${response['status']}');
}
logTrace(actualCorrelationId, 'Uploaded to URL: ${response['url']}');
final payload = _buildPayload(dataname, payloadType, payloadBytes, 'link', response['url']);
payloads.add(payload);
}
}
// Build envelope
final env = _buildEnvelope(subject, payloads, {
'correlation_id': actualCorrelationId,
'msg_id': actualMsgId,
'msg_purpose': msgPurpose,
'sender_name': senderName,
'sender_id': actualSenderId,
'receiver_name': receiverName,
'receiver_id': receiverId,
'reply_to': replyTo,
'reply_to_msg_id': replyToMsgId,
'broker_url': brokerUrl,
});
final envJsonStr = json.encode(env);
if (isPublish) {
if (natsConnection != null) {
await publishMessage(natsConnection, subject, envJsonStr, actualCorrelationId);
} else {
await publishMessage(brokerUrl, subject, envJsonStr, actualCorrelationId);
}
}
return [env, envJsonStr];
}
/// Receive and process NATS message
///
/// This function processes incoming NATS messages, handling both direct transport
/// (base64 decoded payloads) and link transport (URL-based payloads).
/// It deserializes the data based on the transport type and returns the result.
///
/// [msg] - NATS message to process (dict with 'payloads' key)
/// [options] - Optional configuration
///
/// Returns a Future that completes with the envelope object with processed payloads
Future<Map<String, dynamic>> smartreceive(
Map<String, dynamic> msg, {
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
int maxDelay = 5000,
}) async {
final correlationId = msg['correlation_id'] as String;
logTrace(correlationId, 'Processing received message');
// Process all payloads in the envelope
final payloadsList = <List<dynamic>>[];
final numPayloads = (msg['payloads'] as List).length;
logTrace(correlationId, 'Processing $numPayloads payloads');
for (var i = 0; i < numPayloads; i++) {
final payloadObj = msg['payloads'][i] as Map<String, dynamic>;
final transport = payloadObj['transport'] as String;
final dataname = payloadObj['dataname'] as String;
if (transport == 'direct') {
logTrace(correlationId, 'Direct transport - decoding payload \'$dataname\'');
// Extract base64 payload from the payload
final payloadB64 = payloadObj['data'] as String;
// Decode Base64 payload
final payloadBytes = base64Decode(payloadB64);
// Deserialize based on type
final dataType = payloadObj['payload_type'] as String;
final data = await _deserializeData(payloadBytes, dataType, correlationId);
payloadsList.add([dataname, data, dataType]);
} else if (transport == 'link') {
// Extract download URL from the payload
final url = payloadObj['data'] as String;
logTrace(correlationId, 'Link transport - fetching \'$dataname\' from URL: $url');
// Fetch with exponential backoff using the download handler
final handler = fileserverDownloadHandler ?? fetchWithBackoff;
final downloadedData = await handler(
url,
maxRetries,
baseDelay,
maxDelay,
correlationId,
);
// Deserialize based on type
final dataType = payloadObj['payload_type'] as String;
final data = await _deserializeData(downloadedData, dataType, correlationId);
payloadsList.add([dataname, data, dataType]);
} else {
throw Exception('Unknown transport type for payload \'$dataname\': $transport');
}
}
msg['payloads'] = payloadsList;
return msg;
}
// ---------------------------------------------- Module Exports ---------------------------------------------- //
/// Convenience class for NATSBridge functionality
class NATSBridge {
static const DEFAULT_SIZE_THRESHOLD = DEFAULT_SIZE_THRESHOLD;
static const DEFAULT_BROKER_URL = DEFAULT_BROKER_URL;
static const DEFAULT_FILESERVER_URL = DEFAULT_FILESERVER_URL;
/// Send data via NATS
static Future<List<dynamic>> send(
String subject,
List<List<dynamic>> data, {
String brokerUrl = DEFAULT_BROKER_URL,
String fileserverUrl = DEFAULT_FILESERVER_URL,
Function? fileserverUploadHandler,
int sizeThreshold = DEFAULT_SIZE_THRESHOLD,
String? correlationId,
String msgPurpose = 'chat',
String senderName = 'NATSBridge',
String receiverName = '',
String receiverId = '',
String replyTo = '',
String replyToMsgId = '',
bool isPublish = true,
dynamic natsConnection,
String? msgId,
String? senderId,
}) {
return smartsend(
subject,
data,
brokerUrl: brokerUrl,
fileserverUrl: fileserverUrl,
fileserverUploadHandler: fileserverUploadHandler,
sizeThreshold: sizeThreshold,
correlationId: correlationId,
msgPurpose: msgPurpose,
senderName: senderName,
receiverName: receiverName,
receiverId: receiverId,
replyTo: replyTo,
replyToMsgId: replyToMsgId,
isPublish: isPublish,
natsConnection: natsConnection,
msgId: msgId,
senderId: senderId,
);
}
/// Receive and process NATS message
static Future<Map<String, dynamic>> receive(
Map<String, dynamic> msg, {
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
int maxDelay = 5000,
}) {
return smartreceive(
msg,
fileserverDownloadHandler: fileserverDownloadHandler,
maxRetries: maxRetries,
baseDelay: baseDelay,
maxDelay: maxDelay,
);
}
}
// Base64 encoding/decoding utilities
// These functions are re-exported from dart:convert for convenience
// The dart:convert library provides these functions directly
// String base64Encode(Uint8List data) - from dart:convert
// Uint8List base64Decode(String data) - from dart:convert
// Re-export base64 from dart:convert for convenience
export 'dart:convert' show base64Encode, base64Decode;

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,74 @@
/// Dart Mix Payloads Receiver Test
/// Tests the smartreceive function with mixed payload types
///
/// This test mirrors test_julia_mix_payloads_receiver.jl and test_js_mix_payloads_receiver.js
/// and demonstrates that any combination and any number of mixed content can be received correctly.
import 'dart:io';
import 'package:http/http.dart' as http;
import 'package:uuid/uuid.dart';
// Add parent directory to path
import 'package:natsbridge/natsbridge.dart' as natsbridge;
const TEST_SUBJECT = '/natsbridge';
const TEST_BROKER_URL = String.fromEnvironment(
'NATS_URL',
defaultValue: 'nats.yiem.cc',
);
const TEST_FILESERVER_URL = String.fromEnvironment(
'FILESERVER_URL',
defaultValue: 'http://192.168.88.104:8080',
);
void logTrace(String message) {
final timestamp = DateTime.now().toUtc().toIsoString();
print('[$timestamp] [Correlation: $correlationId] $message');
}
Future<void> runTest() async {
print('=== Dart Mix Payloads Receiver Test ===\n');
final uuid = const Uuid();
final correlationId = uuid.v4();
print('Correlation ID: $correlationId');
print('Subject: $TEST_SUBJECT');
print('Broker URL: $TEST_BROKER_URL');
print('Fileserver URL: $TEST_FILESERVER_URL\n');
bool testPassed = true;
int messagesReceived = 0;
final receivedPayloads = [];
try {
// Note: This is a receiver test that waits for messages
// You need to run the sender test first: dart test/test_dart_mix_payloads_sender.dart
print('This receiver test requires a running NATS server and a message sender.');
print('\nTo run this test:');
print('1. Start NATS server: nats-server');
print('2. Run sender: dart test/test_dart_mix_payloads_sender.dart');
print('3. This receiver will wait for messages on subject: $TEST_SUBJECT\n');
print('Waiting for messages (timeout: 180 seconds)...');
// For now, just print a message about how to run the test
// In a real implementation, you would connect to NATS and subscribe to messages
print('\n=== Test Instructions ===');
print('1. Start NATS server: nats-server');
print('2. Run sender: dart test/test_dart_mix_payloads_sender.dart');
print('3. This receiver will wait for messages\n');
print('Test completed. This is a receiver test that waits for messages from the sender.');
print('Run the sender test first to send messages to this receiver.');
} catch (error) {
print('\n❌ Test failed with error: $error');
print('$error');
exit(1);
}
}
void main() {
runTest();
}

View File

@@ -0,0 +1,230 @@
/// Dart Mix Payloads Sender Test
/// Tests the smartsend function with mixed payload types
///
/// This test mirrors test_julia_mix_payloads_sender.jl and test_js_mix_payloads_sender.js
/// and demonstrates that any combination and any number of mixed content can be sent correctly.
import 'dart:io';
import 'dart:typed_data';
import 'package:http/http.dart' as http;
import 'package:uuid/uuid.dart';
// Add parent directory to path
import 'package:natsbridge/natsbridge.dart' as natsbridge;
const TEST_SUBJECT = '/natsbridge';
const TEST_BROKER_URL = String.fromEnvironment(
'NATS_URL',
defaultValue: 'nats.yiem.cc',
);
const TEST_FILESERVER_URL = String.fromEnvironment(
'FILESERVER_URL',
defaultValue: 'http://192.168.88.104:8080',
);
const SIZE_THRESHOLD = 1000000; // 1MB threshold
void logTrace(String message) {
final timestamp = DateTime.now().toUtc().toIsoString();
print('[$timestamp] [Correlation: $correlationId] $message');
}
Future<void> runTest() async {
print('=== Dart Mix Payloads Sender Test ===\n');
final uuid = const Uuid();
final correlationId = uuid.v4();
print('Correlation ID: $correlationId');
print('Subject: $TEST_SUBJECT');
print('Broker URL: $TEST_BROKER_URL');
print('Fileserver URL: $TEST_FILESERVER_URL');
print('Size Threshold: $SIZE_THRESHOLD bytes (1MB)\n');
// Create sample data for each type (mirroring Julia test)
final textData = 'Hello! This is a test chat message. 🎉\nHow are you doing today? 😊';
final dictData = {
'type': 'chat',
'sender': 'serviceA',
'receiver': 'serviceB',
'metadata': {
'timestamp': DateTime.now().toUtc().toIsoString(),
'priority': 'high',
'tags': ['urgent', 'chat', 'test']
},
'content': {
'text': 'This is a JSON-formatted chat message with nested structure.',
'format': 'markdown',
'mentions': ['user1', 'user2']
}
};
// Arrow table data (small - direct transport)
final arrowTableSmall = [
{'id': 1, 'name': 'Alice', 'score': 95, 'active': true},
{'id': 2, 'name': 'Bob', 'score': 88, 'active': false},
{'id': 3, 'name': 'Charlie', 'score': 92, 'active': true},
{'id': 4, 'name': 'Diana', 'score': 78, 'active': true},
{'id': 5, 'name': 'Eve', 'score': 85, 'active': false},
{'id': 6, 'name': 'Frank', 'score': 91, 'active': true},
{'id': 7, 'name': 'Grace', 'score': 89, 'active': true},
{'id': 8, 'name': 'Henry', 'score': 76, 'active': false},
{'id': 9, 'name': 'Ivy', 'score': 94, 'active': true},
{'id': 10, 'name': 'Jack', 'score': 82, 'active': true}
];
// Json table data (small - direct transport)
final jsonTableSmall = [
{'id': 1, 'name': 'Alice', 'score': 95, 'active': true},
{'id': 2, 'name': 'Bob', 'score': 88, 'active': false},
{'id': 3, 'name': 'Charlie', 'score': 92, 'active': true},
{'id': 4, 'name': 'Diana', 'score': 78, 'active': true},
{'id': 5, 'name': 'Eve', 'score': 85, 'active': false},
{'id': 6, 'name': 'Frank', 'score': 91, 'active': true},
{'id': 7, 'name': 'Grace', 'score': 89, 'active': true},
{'id': 8, 'name': 'Henry', 'score': 76, 'active': false},
{'id': 9, 'name': 'Ivy', 'score': 94, 'active': true},
{'id': 10, 'name': 'Jack', 'score': 82, 'active': true}
];
// Audio data (small binary - direct transport)
final audioData = Uint8List(100);
for (int i = 0; i < 100; i++) {
audioData[i] = (Random().nextRange(1, 255)).toInt();
}
// Video data (small binary - direct transport)
final videoData = Uint8List(150);
for (int i = 0; i < 150; i++) {
videoData[i] = (Random().nextRange(1, 255)).toInt();
}
// Binary data (small - direct transport)
final binaryData = Uint8List(200);
for (int i = 0; i < 200; i++) {
binaryData[i] = (Random().nextRange(1, 255)).toInt();
}
// Large data for link transport testing
final largeArrowTable = List.generate(20000, (i) => {
'id': i + 1,
'name': 'user_${i + 1}',
'score': (Random().nextRange(50, 100)).toInt(),
'active': Random().nextBool(),
'timestamp': DateTime.now().toUtc().toIsoString()
});
final largeJsonTable = List.generate(50000, (i) => {
'id': i + 1,
'name': 'user_${i + 1}',
'score': (Random().nextRange(50, 100)).toInt(),
'active': Random().nextBool()
});
final largeAudioData = Uint8List(1500000);
for (int i = 0; i < 1500000; i++) {
largeAudioData[i] = (Random().nextRange(1, 255)).toInt();
}
final largeVideoData = Uint8List(1500000);
for (int i = 0; i < 1500000; i++) {
largeVideoData[i] = (Random().nextRange(1, 255)).toInt();
}
final largeBinaryData = Uint8List(1500000);
for (int i = 0; i < 1500000; i++) {
largeBinaryData[i] = (Random().nextRange(1, 255)).toInt();
}
// Read image files from disk (following Julia test pattern)
final filePathSmallImage = './test/small_image.jpg';
final fileDataSmallImage = File(filePathSmallImage).readAsBytesSync();
final filenameSmallImage = filePathSmallImage.split('/').last;
final filePathLargeImage = './test/large_image.png';
final fileDataLargeImage = File(filePathLargeImage).readAsBytesSync();
final filenameLargeImage = filePathLargeImage.split('/').last;
logTrace('Creating payloads list with mixed content');
// Create payloads list - mixed content with both small and large data
// Small data uses direct transport, large data uses link transport
final payloads = [
// Small data (direct transport) - text, dictionary, arrowtable, jsontable, small image
['chat_text', textData, 'text'],
['chat_json', dictData, 'dictionary'],
// ['arrow_table_small', arrowTableSmall, 'arrowtable'],
['json_table_small', jsonTableSmall, 'jsontable'],
[filenameSmallImage, fileDataSmallImage, 'binary'],
// Large data (link transport) - large arrowtable, large jsontable, large image, large audio, large video, large binary
// ['arrow_table_large', largeArrowTable, 'arrowtable'],
['json_table_large', largeJsonTable, 'jsontable'],
[filenameLargeImage, fileDataLargeImage, 'binary'],
// ['audio_clip_large', largeAudioData, 'audio'],
// ['video_clip_large', largeVideoData, 'video'],
// ['binary_file_large', largeBinaryData, 'binary']
];
logTrace('Total payloads: ${payloads.length}');
try {
// Send the message
print('Sending mixed payloads...\n');
final (env, envJsonStr) = await natsbridge.smartsend(
TEST_SUBJECT,
payloads,
brokerUrl: TEST_BROKER_URL,
fileserverUrl: TEST_FILESERVER_URL,
fileserverUploadHandler: natsbridge.plikOneshotUpload,
sizeThreshold: SIZE_THRESHOLD,
correlationId: correlationId,
msgPurpose: 'chat',
senderName: 'dart-mix-test',
receiverName: '',
receiverId: '',
replyTo: '',
replyToMsgId: '',
isPublish: true,
);
print('\n=== Envelope Created ===');
print('Correlation ID: ${env['correlation_id']}');
print('Message ID: ${env['msg_id']}');
print('Timestamp: ${env['timestamp']}');
print('Subject: ${env['send_to']}');
print('Purpose: ${env['msg_purpose']}');
print('Sender: ${env['sender_name']}');
print('Payloads: ${env['payloads'].length}\n');
// Log transport type for each payload
for (int i = 0; i < env['payloads'].length; i++) {
final payload = env['payloads'][i];
logTrace('Payload ${i + 1} (${payload['dataname']}):');
logTrace(' Transport: ${payload['transport']}');
logTrace(' Type: ${payload['payload_type']}');
logTrace(' Size: ${payload['size']} bytes');
logTrace(' Encoding: ${payload['encoding']}');
if (payload['transport'] == 'link') {
logTrace(' URL: ${payload['data']}');
}
}
// Summary
print('\n--- Transport Summary ---');
final directCount = env['payloads'].where((p) => p['transport'] == 'direct').length;
final linkCount = env['payloads'].where((p) => p['transport'] == 'link').length;
logTrace('Direct transport: $directCount payloads');
logTrace('Link transport: $linkCount payloads');
print('\nTest completed.');
} catch (error) {
print('\n❌ Test failed with error: $error');
print('$error');
exit(1);
}
}
void main() {
runTest();
}