8 Commits

Author SHA1 Message Date
ton
6da64092ca Merge pull request 'v0.6.0-dev-seperate_natsclient_smartsend' (#13) from v0.6.0-dev-seperate_natsclient_smartsend into v0.6.0-dev
Reviewed-on: #13
2026-05-14 06:21:29 +00:00
809aea454b update 2026-05-14 13:16:13 +07:00
c5a70edd57 rust version implemented 2026-05-13 20:24:08 +07:00
b0acee053c update docs 2026-05-13 17:35:46 +07:00
c25c6a8a43 update 2026-05-13 16:57:20 +07:00
34a6d19303 update jl docstring 2026-05-13 16:25:48 +07:00
8ada1ca49c update 2026-05-13 16:08:29 +07:00
60ae464ea2 update 2026-05-13 16:02:50 +07:00
18 changed files with 4029 additions and 1635 deletions

1
.gitignore vendored
View File

@@ -1,3 +1,4 @@
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 --------------------------------------------- #
Check NATSBridge/docs folder I want to update the content of the following files according to ASG_Framework/ASG_Framework.md:
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:
- NATSBridge/docs/requirements.md
- NATSBridge/docs/specification.md
- NATSBridge/docs/ui-specification.md (you'll need to create this one)
@@ -153,20 +153,50 @@ I'll do the other docs not listed here later myself.
now help me update the following fileaccording to ASG_Framework/ASG_Framework.md:
now help me update the following file according to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/specification.md
<!-- ------------------------------------------- 100 ------------------------------------------- -->
Check NATSBridge/docs folder. I would like to expand this package to include Dart support.
Can you update the content of the following files according to ASG_Framework/ASG_Framework.md:
- NATSBridge/docs/requirements.md
- NATSBridge/docs/specification.md
- NATSBridge/docs/walkthrough.md
- NATSBridge/docs/architecture.md
Check ./docs folder. I would like to expand this package (NATSBRIDGE) to include Rust support.
Can you update the content of the following files according to /home/ton/docker-apps/sommpanion/ASG_Framework/ASG_Framework.md:
- ./docs/requirements.md
- ./docs/specification.md
- ./docs/walkthrough.md
- ./docs/architecture.md
<!-- ------------------------------------------- 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 Normal file

File diff suppressed because it is too large Load Diff

31
Cargo.toml Normal file
View File

@@ -0,0 +1,31 @@
[package]
name = "natsbridge"
version = "1.2.0"
edition = "2021"
description = "Cross-platform bi-directional data bridge for NATS communication"
[lib]
name = "natsbridge"
path = "src/natsbridge.rs"
[dependencies]
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["json", "stream", "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 = "8d8e0b0f350b8e1c91420b5e64e5de774c2f0f4d"
git-tree-sha1 = "deddd8725e5e1cc49ee205a1964256043720a6c3"
uuid = "336ed68f-0bac-5ca0-87d4-7b16caf5d00b"
version = "0.10.16"
version = "0.10.15"
[[deps.CodeTracking]]
deps = ["InteractiveUtils", "UUIDs"]
@@ -108,9 +108,9 @@ version = "1.3.0+1"
[[deps.ConcurrentUtilities]]
deps = ["Serialization", "Sockets"]
git-tree-sha1 = "21d088c496ea22914fe80906eb5bce65755e5ec8"
git-tree-sha1 = "d9d26935a0bcffc87d2613ce14c527c99fc543fd"
uuid = "f0e56b4a-5159-44fe-b623-3e5288b988bb"
version = "2.5.1"
version = "2.5.0"
[[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 = "c49898e8438c828577f04b92fc9368c388ac783c"
git-tree-sha1 = "7bebc8aad6ee6217c78c5ddcf7ed289d65d0263e"
uuid = "4e289a0a-7415-4d19-859d-a7e5c4648b56"
version = "1.0.7"
version = "1.0.6"
[[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 = "51059d23c8bb67911a2e6fd5130229113735fc7e"
git-tree-sha1 = "5e6fe50ae7f23d171f44e311c2960294aaa0beb5"
uuid = "cd3eb016-35fb-5094-929b-558a96fad6f3"
version = "1.11.0"
version = "1.10.19"
[[deps.HashArrayMappedTries]]
git-tree-sha1 = "2eaa69a7cab70a52b9687c8bf950a5a93ec895ae"
@@ -307,9 +307,9 @@ weakdeps = ["ArrowTypes"]
[[deps.JuliaInterpreter]]
deps = ["CodeTracking", "InteractiveUtils", "Random", "UUIDs"]
git-tree-sha1 = "3d3b79166e2a0afcf875df20db110af91ad3ab61"
git-tree-sha1 = "80580012d4ed5a3e8b18c7cd86cebe4b816d17a6"
uuid = "aa1ae85d-cabe-5617-a682-6adf51b2e16a"
version = "0.10.11"
version = "0.10.9"
[[deps.JuliaSyntaxHighlighting]]
deps = ["StyledStrings"]
@@ -383,9 +383,9 @@ version = "1.2.0"
[[deps.LoweredCodeUtils]]
deps = ["CodeTracking", "Compiler", "JuliaInterpreter"]
git-tree-sha1 = "5d4278f755440f70648d80cc6225f51e78e94094"
git-tree-sha1 = "65ae3db6ab0e5b1b5f217043c558d9d1d33cc88d"
uuid = "6f1432cf-f94c-5a45-995e-cdbf5db27b0b"
version = "3.5.1"
version = "3.5.0"
[[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 = "8785729fa736197687541f7053f6d8ab7fc44f92"
git-tree-sha1 = "c067a280ddc25f196b5e7df3877c6b226d390aaf"
uuid = "739be429-bea8-5141-9913-cc70e7f3736d"
version = "1.1.10"
version = "1.1.9"
[[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", "Base64", "DataFrames", "Dates", "GeneralUtils", "HTTP", "JSON", "NATS", "PrettyPrinting", "Revise", "UUIDs"]
deps = ["Arrow", "DataFrames", "Dates", "GeneralUtils", "HTTP", "JSON", "NATS", "PrettyPrinting", "Revise", "UUIDs"]
path = "."
uuid = "f2724d33-f338-4a57-b9f8-1be882570d10"
version = "0.5.6"
version = "0.4.1"
[[deps.NanoDates]]
deps = ["Dates", "Parsers"]
@@ -514,9 +514,9 @@ version = "1.3.3"
[[deps.Preferences]]
deps = ["TOML"]
git-tree-sha1 = "8b770b60760d4451834fe79dd483e318eee709c4"
git-tree-sha1 = "522f093a29b31a93e34eaea17ba055d850edea28"
uuid = "21216c6a-2e73-6563-6e65-726566657250"
version = "1.5.2"
version = "1.5.1"
[[deps.PrettyPrinting]]
git-tree-sha1 = "142ee93724a9c5d04d78df7006670a93ed1b244e"
@@ -525,15 +525,9 @@ version = "0.4.2"
[[deps.PrettyTables]]
deps = ["Crayons", "LaTeXStrings", "Markdown", "PrecompileTools", "Printf", "REPL", "Reexport", "StringManipulation", "Tables"]
git-tree-sha1 = "211530a7dc76ab59087f4d4d1fc3f086fbe87594"
git-tree-sha1 = "c5a07210bd060d6a8491b0ccdee2fa0235fc00bf"
uuid = "08abe8d2-0d0c-5749-adfa-8a2ac140af0d"
version = "3.2.3"
[deps.PrettyTables.extensions]
PrettyTablesTypstryExt = "Typstry"
[deps.PrettyTables.weakdeps]
Typstry = "f0ed7684-a786-439e-b1e3-3b82803b501e"
version = "3.1.2"
[[deps.Printf]]
deps = ["Unicode"]
@@ -541,9 +535,9 @@ uuid = "de0858da-6303-5e67-8744-51eddeeeb8d7"
version = "1.11.0"
[[deps.PtrArrays]]
git-tree-sha1 = "4fbbafbc6251b883f4d2705356f3641f3652a7fe"
git-tree-sha1 = "1d36ef11a9aaf1e8b74dacc6a731dd1de8fd493d"
uuid = "43287f4e-b6f4-7ad1-bb20-aadabca52c3d"
version = "1.4.0"
version = "1.3.0"
[[deps.QuadGK]]
deps = ["DataStructures", "LinearAlgebra"]
@@ -574,9 +568,9 @@ version = "1.2.2"
[[deps.Revise]]
deps = ["CodeTracking", "FileWatching", "InteractiveUtils", "JuliaInterpreter", "LibGit2", "LoweredCodeUtils", "OrderedCollections", "Preferences", "REPL", "UUIDs"]
git-tree-sha1 = "d97d78d4fc5f858d8ce44f6b88bc972f2023f51d"
git-tree-sha1 = "14d1bfb0a30317edc77e11094607ace3c800f193"
uuid = "295af30f-e4ad-537b-8983-00126c2a3abe"
version = "3.14.0"
version = "3.13.2"
[deps.Revise.extensions]
DistributedExt = "Distributed"
@@ -602,9 +596,9 @@ version = "0.7.0"
[[deps.ScopedValues]]
deps = ["HashArrayMappedTries", "Logging"]
git-tree-sha1 = "ac4b837d89a58c848e85e698e2a2514e9d59d8f6"
git-tree-sha1 = "c3b2323466378a2ba15bea4b2f73b081e022f473"
uuid = "7e506255-f358-4e82-b7e4-beb19740aa63"
version = "1.6.0"
version = "1.5.0"
[[deps.Scratch]]
deps = ["Dates"]
@@ -650,9 +644,9 @@ version = "1.12.0"
[[deps.SpecialFunctions]]
deps = ["IrrationalConstants", "LogExpFunctions", "OpenLibm_jll", "OpenSpecFun_jll"]
git-tree-sha1 = "5acc6a41b3082920f79ca3c759acbcecf18a8d78"
git-tree-sha1 = "f2685b435df2613e25fc10ad8c26dddb8640f547"
uuid = "276daf66-3868-5448-9aa4-cd146d93841b"
version = "2.7.1"
version = "2.6.1"
[deps.SpecialFunctions.extensions]
SpecialFunctionsChainRulesCoreExt = "ChainRulesCore"
@@ -698,9 +692,9 @@ version = "1.5.2"
[[deps.StringManipulation]]
deps = ["PrecompileTools"]
git-tree-sha1 = "d05693d339e37d6ab134c5ab53c29fce5ee5d7d5"
git-tree-sha1 = "a3c1536470bf8c5e02096ad4853606d7c8f62721"
uuid = "892a3eda-7b42-436c-8928-eab12a02cf0e"
version = "0.4.4"
version = "0.4.2"
[[deps.StringViews]]
git-tree-sha1 = "f2dcb92855b31ad92fe8f079d4f75ac57c93e4b8"
@@ -715,18 +709,16 @@ version = "1.11.0"
[[deps.StructUtils]]
deps = ["Dates", "UUIDs"]
git-tree-sha1 = "fa95b3b097bcef5845c142ea2e085f1b2591e92c"
git-tree-sha1 = "9297459be9e338e546f5c4bedb59b3b5674da7f1"
uuid = "ec057cc2-7a8d-4b58-b3b3-92acb9f63b42"
version = "2.7.1"
version = "2.6.2"
[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**, **Dart**, 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**, 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,35 +48,33 @@ 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 | 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 | ❌ | ❌ | ✅ | ❌ | ✅ | ✅ | ❌ |
| 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 | ❌ | ❌ | ✅ | ❌ | ❌ |
---
## Features
-**Cross-platform messaging** for Julia, JavaScript, Python, Dart, and MicroPython applications
-**Cross-platform messaging** for Julia, JavaScript, Python, 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/Dart)
-**Apache Arrow IPC** support for tabular data (Desktop: Julia/Python/Node.js)
-**JSON Table** support for tabular data (All platforms including Browser)
-**Exponential backoff** for reliable file server downloads
-**Correlation ID tracking** for message tracing
@@ -173,38 +171,6 @@ 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
@@ -369,54 +335,6 @@ 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.
@@ -500,50 +418,20 @@ 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 | 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 |
| 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 |
---
@@ -833,7 +721,6 @@ 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
@@ -901,30 +788,6 @@ 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.1.0
**Date**: 2026-03-23
**Version**: 1.2.0
**Date**: 2026-05-13
**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**, and **MicroPython** applications using NATS as the message bus.
This document defines the **blueprint** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This architecture document serves as the single source of truth for:
- **System Structure**: How components fit together and interact
@@ -57,6 +57,7 @@ flowchart TD
JS_App[JavaScript Application<br/>Node.js/Browser]
Python_App[Python Application<br/>Desktop]
Dart_App[Dart Application<br/>Desktop/Flutter/Web]
Rust_App[Rust Application<br/>Server/Desktop]
MicroPython_App[MicroPython Device]
end
@@ -64,12 +65,14 @@ flowchart TD
JS_App -->|NATS| NATS_Server
Python_App -->|NATS| NATS_Server
Dart_App -->|NATS| NATS_Server
Rust_App -->|NATS| NATS_Server
MicroPython_App -->|NATS| NATS_Server
Julia_App -->|HTTP| File_Server
JS_App -->|HTTP| File_Server
Python_App -->|HTTP| File_Server
Dart_App -->|HTTP| File_Server
Rust_App -->|HTTP| File_Server
MicroPython_App -->|HTTP| File_Server
style NATS_Server fill:#fff3e0,stroke:#f57c00
@@ -78,6 +81,7 @@ flowchart TD
style JS_App fill:#e3f2fd,stroke:#2196f3
style Python_App fill:#e3f2fd,stroke:#2196f3
style Dart_App fill:#fff0f6,stroke:#e91e63
style Rust_App fill:#dea584,stroke:#e65100
style MicroPython_App fill:#fce4ec,stroke:#e91e63
```
@@ -85,28 +89,20 @@ 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
@@ -115,6 +111,7 @@ flowchart TD
JS_Module --> File_Client
Python_Module --> File_Client
Dart_Module --> File_Client
Rust_Module --> File_Client
MicroPython_Module --> File_Client
File_Client --> File_Server
@@ -123,6 +120,7 @@ flowchart TD
style JS_Module fill:#e3f2fd,stroke:#2196f3
style Python_Module fill:#e3f2fd,stroke:#2196f3
style Dart_Module fill:#fff0f6,stroke:#e91e63
style Rust_Module fill:#dea584,stroke:#e65100
style MicroPython_Module fill:#fce4ec,stroke:#e91e63
style NATS_Broker fill:#fff3e0,stroke:#f57c00
style File_Server fill:#f3e5f5,stroke:#9c27b4
@@ -139,36 +137,31 @@ flowchart TD
Serialize[_serialize_data]
Deserialize[_deserialize_data]
BuildEnvelope[build_envelope]
BuildPayload[build_payload]
PublishMessage[publish_message]
EnvelopeToJson[envelope_to_json]
FileServerUpload[fileserver_upload_handler]
FileServerDownload[fileserver_download_handler]
LogTrace[log_trace]
end
subgraph "Data Models"
Payload[MsgPayloadV1 Struct]
Envelope[MsgEnvelopeV1 Struct]
Payload[msg_payload_v1 Struct]
Envelope[msg_envelope_v1 Struct]
end
SmartSend --> Serialize
SmartSend --> BuildEnvelope
SmartSend --> BuildPayload
SmartSend --> PublishMessage
SmartSend --> EnvelopeToJson
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
```
@@ -181,15 +174,14 @@ flowchart TD
| Component | Purpose | Platform Support |
|-----------|---------|------------------|
| **smartsend** | Send data via NATS with automatic transport selection | All |
| **smartreceive** | Receive and process NATS messages | All |
| **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 |
| **_serialize_data** | Serialize data according to payload type | All |
| **_deserialize_data** | Deserialize bytes to native data types | All |
| **_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) |
| **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) |
### Data Flow
@@ -211,7 +203,7 @@ flowchart TD
H --> L[Build envelope]
L --> M[Convert to JSON]
M --> N[Publish to NATS]
M --> N[Return envelope + JSON to caller]
style A fill:#f9f9f9,stroke:#333
style N fill:#e0e7ff,stroke:#3b82f6
@@ -304,7 +296,7 @@ end
|------|-------------|---------------|----------|-----------|
| `text` | Plain text string | UTF-8 bytes | Base64 | All |
| `dictionary` | JSON object | JSON string | Base64/JSON | All |
| `arrowtable` | Apache Arrow IPC | Arrow IPC stream | Base64/arrow-ipc | Desktop (Julia/Python/Node.js/Dart) |
| `arrowtable` | Apache Arrow IPC | Arrow IPC stream | Base64/arrow-ipc | Desktop (Julia/Python/Node.js/Dart/Rust) |
| `jsontable` | JSON array of objects | JSON string | Base64/json | All (including Browser/Dart Web) |
| `image` | Binary image data | Raw bytes | Base64 | All |
| `audio` | Binary audio data | Raw bytes | Base64 | All |
@@ -437,11 +429,9 @@ 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)
@@ -449,36 +439,6 @@ 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`
@@ -486,23 +446,6 @@ class NATSConnectionPool {
- **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:
@@ -590,6 +533,62 @@ DEFAULT_SIZE_THRESHOLD = 100_000 # 100KB
MAX_PAYLOAD_SIZE = 50_000 # 50KB hard limit
```
### Rust Architecture
Rust leverages compile-time type safety and async runtimes:
- **Type-safe payloads**: Rust enum discriminates between `Text`, `Dictionary`, `ArrowTable`, `Binary`, etc.
- **serde serialization**: Automatic JSON deserialization via `#[derive(Serialize, Deserialize)]`
- **tokio runtime**: Efficient async I/O for NATS connections and HTTP file server operations
- **arrow2 integration**: Native Arrow IPC deserialization without intermediate format conversion
- **reqwest**: High-performance HTTP client with built-in TLS and connection pooling
- **Zero-copy patterns**: `Vec<u8>` passed directly to avoid unnecessary memory copies
- **Result<T, E>**: Idiomatic error handling with typed error types
```rust
// Type-safe payload enum (compile-time discrimination)
#[derive(Serialize, Deserialize, Clone)]
pub enum Payload {
Text(String),
Dictionary(serde_json::Value),
ArrowTable(Vec<u8>),
JsonTable(serde_json::Value),
Image(Vec<u8>),
Audio(Vec<u8>),
Video(Vec<u8>),
Binary(Vec<u8>),
}
// Configuration via builder pattern
pub struct SmartsendOptions {
pub broker_url: String,
pub fileserver_url: String,
pub fileserver_upload_handler: Option<UploadHandler>,
pub size_threshold: usize,
pub correlation_id: String,
pub msg_purpose: String,
pub sender_name: String,
// ... other fields
}
// NATS client with tokio integration
let conn = nats::connect("nats://localhost:4222").await?;
// Subscribe and process messages
let mut sub = conn.subscribe("/agent/wine/api/v1/analyze")?;
for msg in sub.messages() {
let envelope: MsgEnvelopeV1 = serde_json::from_slice(&msg.payload)?;
// Type-safe access to payloads
for payload in &envelope.payloads {
match &payload.data {
Payload::ArrowTable(bytes) => { /* process */ },
Payload::Text(text) => { /* process */ },
_ => {}
}
}
}
```
---
## Scaling Architecture
@@ -740,7 +739,7 @@ MAX_PAYLOAD_SIZE = 50_000 # 50KB hard limit
|----------|---------|-------------|
| `NATS_URL` | `nats://localhost:4222` | NATS server URL |
| `FILESERVER_URL` | `http://localhost:8080` | HTTP file server URL |
| `SIZE_THRESHOLD` | `1000000` | Size threshold in bytes |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
### Container Deployment
@@ -826,7 +825,7 @@ flowchart TD
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, MicroPython 1.19+ |
---
@@ -834,6 +833,16 @@ flowchart TD
| Date | Version | Changes |
|------|---------|---------|
| 2026-05-13 | 1.3.0 | Added Rust support with tokio, serde, and arrow2 | All sections |
| - | - | Added Rust to C4 diagrams (context, container) | All sections |
| - | - | Added Rust platform-specific architecture section | specification.md:13 |
| - | - | Updated component table with Rust support | All sections |
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) |
| - | - | Removed publish_message component (commented out in source) |
| - | - | Removed NATSClient and NATSConnectionPool classes (not in ground truth) |
| - | - | 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 |
@@ -865,6 +874,7 @@ flowchart TD
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | specification.md:2-19 (all sections) | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | specification.md:2-19 (all sections) | FR-005, FR-006, FR-012 |
### 16.3 External Dependencies
@@ -887,6 +897,13 @@ flowchart TD
| Dart | http | Latest | HTTP file server | specification.md:11 | FR-008, FR-009 |
| Dart | uuid | Latest | UUID generation | specification.md:11 | FR-011, NFR-401 |
| Dart | dart-arrow | Latest | Arrow IPC support | specification.md:11 | FR-002, FR-012 |
| Rust | nats | Latest | NATS client | specification.md:11 | FR-013, FR-014 |
| Rust | serde | Latest | JSON serialization | specification.md:11 | FR-012, NFR-101, NFR-102 |
| Rust | serde_json | Latest | JSON handling | specification.md:11 | FR-012, NFR-101, NFR-102 |
| Rust | tokio | Latest | Async runtime | specification.md:11 | FR-013, FR-014 |
| Rust | reqwest | Latest | HTTP file server | specification.md:11 | FR-008, FR-009 |
| Rust | uuid | Latest | UUID generation | specification.md:11 | FR-011, NFR-401 |
| Rust | arrow2 | Latest | Arrow IPC support | specification.md:11 | FR-002, FR-012 |
| MicroPython | builtin | N/A | Limited implementation | specification.md:11 | FR-005, FR-006, FR-012 |
---

View File

@@ -1,7 +1,7 @@
# Requirements Document: NATSBridge
**Version**: 1.0.0
**Date**: 2026-03-23
**Version**: 1.2.0
**Date**: 2026-05-13
**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**, and **MicroPython** applications using NATS as the message bus. The system implements the **Claim-Check pattern** for efficient handling of large payloads (>0.5MB) by uploading them to an HTTP file server instead of sending raw binary data over NATS.
NATSBridge is a cross-platform, bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus. The system implements the **Claim-Check pattern** for efficient handling of large payloads (>0.5MB) by uploading them to an HTTP file server instead of sending raw binary data over NATS.
### 1.2 User Stories (with acceptance criteria)
@@ -25,6 +25,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| **As a Dart developer**, I want to send tabular data (List<Map>) to other platforms | P1 | JSON table format exchange works with Arrow IPC on desktop |
| **As a Dart developer**, I want to send large files (>0.5MB) | P1 | Large files are automatically uploaded to file server and URLs are sent via NATS |
| **As a MicroPython developer**, I want to send sensor data with minimal memory usage | P1 | Direct transport works for payloads <100KB on memory-constrained devices |
| **As a Rust developer**, I want to send and receive messages with type-safe APIs | P1 | Rust implementation uses serde for serialization, tokio for async, and nats-io for NATS connectivity |
| **As a developer**, I want to send mixed-content messages (text + image + file) | P1 | NATSBridge accepts list of (dataname, data, type) tuples and handles each payload appropriately |
| **As a developer**, I want to receive multi-payload messages | P1 | NATSBridge returns payloads as list of tuples with correct types preserved |
| **As a developer**, I want to use Plik as the file server | P2 | Plik one-shot upload mode is supported with upload ID and token handling |
@@ -51,7 +52,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Feature | Description |
|---------|-------------|
| Cross-platform interoperability | Seamless data exchange between Julia, JavaScript, Python, Dart, and MicroPython |
| Cross-platform interoperability | Seamless data exchange between Julia, JavaScript, Python, Dart, Rust, and MicroPython |
| Intelligent transport selection | Direct transport (<0.5MB) vs Link transport (≥0.5MB) based on payload size |
| Unified API | Consistent `smartsend()` and `smartreceive()` functions across all platforms |
| Multi-payload support | List of (dataname, data, type) tuples with appropriate handling |
@@ -88,6 +89,11 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart | nats | Latest stable |
| Dart | http | Latest stable |
| Dart | uuid | Latest stable |
| Rust | nats | Latest stable |
| Rust | serde | Latest stable |
| Rust | serde_json | Latest stable |
| Rust | tokio | Latest stable |
| Rust | uuid | Latest stable |
### 2.4 Platform Compatibility
@@ -98,6 +104,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Python | 3.8+ | pyarrow required for arrowtable support |
| Browser | Latest | No Arrow IPC (uses jsontable only) |
| Dart | 2.17+ | Supports Desktop (Dart SDK), Flutter (Dart SDK), and Web (Dart SDK) |
| Rust | 1.70+ | Full support with async/await, Arrow IPC on desktop |
| MicroPython | 1.19+ | Limited to direct transport |
---
@@ -118,8 +125,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 publish messages to NATS subjects |
| **FR-014** | NATS subscription | System shall receive and process NATS messages |
| **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 |
---
@@ -189,14 +196,14 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Type | Julia | JavaScript | Python | Dart | MicroPython | Description |
|------|-------|------------|--------|------|-------------|-------------|
| `text` | `String` | `string` | `str` | `String` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `Map` | `dict` | JSON-serializable data |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | `List<Map>` (Desktop), `List<dynamic>` (Flutter) | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | `List<Map>` | ⚠️ | Tabular data (JSON) - **Only table type in Browser** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Image binary data |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Audio binary data |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `bytearray` | Video binary data |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `Uint8List` | `bytearray` | Generic binary data |
| `text` | `String` | `string` | `str` | `String` | `String` | `str` | Plain text strings |
| `dictionary` | `Dict`, `NamedTuple` | `Object`, `Array` | `dict`, `list` | `Map`, `serde_json::Value` | `String` | `dict` | JSON-serializable data |
| `arrowtable` | `DataFrame`, `Arrow.Table` | ❌ (Browser), ✅ (Node.js) | `pandas.DataFrame` | `List<Map>` (Desktop), `List<dynamic>` (Flutter) | `arrow2::Table` | ❌ | Tabular data (Arrow IPC) |
| `jsontable` | `Vector{NamedTuple}` | `Array<Object>` | `list[dict]` | `Vec<Map>` | ⚠️ | Tabular data (JSON) - **Only table type in Browser** |
| `image` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Image binary data |
| `audio` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Audio binary data |
| `video` | `Vector{UInt8}` | `Uint8Array`, `Buffer` | `bytes` | `Uint8List` | `Vec<u8>` | `bytearray` | Video binary data |
| `binary` | `Vector{UInt8}`, `IOBuffer` | `Uint8Array`, `Buffer` | `bytes`, `bytearray` | `Uint8List` | `Vec<u8>` | `bytearray` | Generic binary data |
### 6.2 Encoding Requirements
@@ -220,6 +227,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart Desktop | 0.5MB | Default size threshold |
| Dart Flutter | 0.5MB | Default size threshold |
| Dart Web | 0.5MB | Default size threshold |
| Rust | 0.5MB | Default size threshold |
| MicroPython | 100KB | Lower threshold for memory constraints |
### 7.2 Maximum Payload Size
@@ -230,6 +238,7 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
| Dart Desktop | Unlimited | Limited by NATS server configuration |
| Dart Flutter | Unlimited | Limited by NATS server configuration |
| Dart Web | Unlimited | Limited by NATS server configuration |
| Rust | Unlimited | Limited by NATS server configuration |
| MicroPython | 50KB | Hard limit due to 256KB-1MB memory |
---
@@ -324,11 +333,11 @@ NATSBridge is a cross-platform, bi-directional data bridge that enables seamless
```julia
function smartsend(
subject::String,
data::AbstractArray{Tuple{String, Any, String}};
broker_url::String = "nats://localhost:4222",
fileserver_url::String = "http://localhost:8080",
data::AbstractArray{Tuple{String, T1, String}, 1};
broker_url::String = DEFAULT_BROKER_URL,
fileserver_url::String = DEFAULT_FILESERVER_URL,
fileserver_upload_handler::Function = plik_oneshot_upload,
size_threshold::Int = 1_000_000,
size_threshold::Int = DEFAULT_SIZE_THRESHOLD,
correlation_id::String = string(uuid4()),
msg_purpose::String = "chat",
sender_name::String = "NATSBridge",
@@ -336,18 +345,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}
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
```
**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::NATS.Msg;
msg_json_str::String;
fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5,
base_delay::Int = 100,
@@ -355,6 +364,8 @@ function smartreceive(
)::JSON.Object{String, Any}
```
**Note**: Pass `String(nats_msg.payload)` from NATS subscription to `smartreceive`.
---
## 12. Deployment Requirements
@@ -374,7 +385,7 @@ function smartreceive(
|----------|---------|-------------|
| `NATS_URL` | `nats://localhost:4222` | NATS server URL |
| `FILESERVER_URL` | `http://localhost:8080` | HTTP file server URL |
| `SIZE_THRESHOLD` | `1000000` | Size threshold in bytes |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
---
@@ -390,7 +401,7 @@ function smartreceive(
| Version | Supported Platforms |
|---------|---------------------|
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Browser (latest), MicroPython 1.19+ |
| v1.0.x | Julia 1.7+, Node.js 16+, Python 3.8+, Dart 2.17+, Rust 1.70+, Browser (latest), MicroPython 1.19+ |
---
@@ -398,6 +409,13 @@ 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 |
---
@@ -410,6 +428,7 @@ function smartreceive(
- [`src/natsbridge.py`](../src/natsbridge.py) - Python implementation
- [`src/natsbridge.dart`](../src/natsbridge.dart) - Dart implementation
- [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) - MicroPython implementation
- [`src/natsbridge.rs`](../src/natsbridge.rs) - Rust implementation
- [`README.md`](../README.md) - Project overview
- [`docs/specification.md`](./specification.md) - Technical specification
- [`docs/ui-specification.md`](./ui-specification.md) - UI specification

View File

@@ -1,7 +1,7 @@
# Specification: NATSBridge
**Version**: 1.1.0
**Date**: 2026-03-23
**Version**: 1.2.0
**Date**: 2026-05-13
**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**, and **MicroPython** applications using NATS as the message bus.
This document defines the **technical contract** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This specification serves as the single source of truth for:
- **Inputs**: What data structures are accepted by `smartsend()`
@@ -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, Any, String}};
broker_url::String = "nats://localhost:4222",
fileserver_url::String = "http://localhost:8080",
data::AbstractArray{Tuple{String, T1, String}, 1};
broker_url::String = DEFAULT_BROKER_URL,
fileserver_url::String = DEFAULT_FILESERVER_URL,
fileserver_upload_handler::Function = plik_oneshot_upload,
size_threshold::Int = 500_000,
size_threshold::Int = DEFAULT_SIZE_THRESHOLD,
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}
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
```
**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,51 +505,27 @@ 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]>;
// 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>;
```
**Note**: NATS publishing is the caller's responsibility.
#### MicroPython
```python
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
@@ -567,12 +543,11 @@ 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
}
```
@@ -593,22 +568,82 @@ 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::NATS.Msg;
msg_json_str::String; # Pass String(nats_msg.payload) from NATS subscription
fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5,
base_delay::Int = 100,
@@ -616,11 +651,13 @@ 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: Any,
msg_json_str: str, # JSON string from NATS message payload
fileserver_download_handler: Callable = fetch_with_backoff,
max_retries: int = 5,
base_delay: int = 100,
@@ -628,11 +665,13 @@ async def smartreceive(
) -> Dict[str, Any]:
```
**Note**: Input is JSON string from NATS message payload.
#### JavaScript (Node.js)
```typescript
async function smartreceive(
msg: Object,
msg_json_str: string, // JSON string from NATS message payload
options?: {
fileserver_download_handler?: Function;
max_retries?: number;
@@ -646,7 +685,7 @@ async function smartreceive(
```typescript
async function smartreceive(
msg: Object,
msg_json_str: string, // JSON string from NATS message payload
options?: {
fileserver_download_handler?: Function;
max_retries?: number;
@@ -656,17 +695,22 @@ async function smartreceive(
): Promise<Object>;
```
**Note**: Input is JSON string from NATS message payload.
#### MicroPython
```python
def smartreceive(msg: Any, **kwargs) -> Dict[str, Any]:
def smartreceive(msg_json_str: str, **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, {
Map<String, dynamic> msg_json_str, // JSON object from NATS message payload
{
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
@@ -680,7 +724,8 @@ Future<Map<String, dynamic>> smartreceive(
```dart
Future<Map<String, dynamic>> smartreceive(
Map<String, dynamic> msg, {
Map<String, dynamic> msg_json_str, // JSON object from NATS message payload
{
Function? fileserverDownloadHandler,
int maxRetries = 5,
int baseDelay = 100,
@@ -690,6 +735,25 @@ Future<Map<String, dynamic>> smartreceive(
}
```
#### Rust
```rust
pub async fn smartreceive(
msg_json_str: &str, // JSON string from NATS message payload
options: &SmartreceiveOptions,
) -> Result<MsgEnvelopeV1, NatSBridgeError>
// SmartreceiveOptions struct
pub struct SmartreceiveOptions {
pub fileserver_download_handler: Option<DownloadHandler>,
pub max_retries: u32,
pub base_delay: u64,
pub max_delay: u64,
}
```
**Note**: Input is JSON string from NATS message payload. Returns `Result<MsgEnvelopeV1, NatSBridgeError>`.
---
## File Server Interface
@@ -703,6 +767,12 @@ 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**:
@@ -796,6 +866,18 @@ function fileserver_download_handler(
| File server download | ✅ Supported | HTTP/HTTPS |
| Size threshold | 500KB | Configurable |
### Rust
| Feature | Status | Notes |
|---------|--------|-------|
| Arrow IPC | ✅ Supported | Requires `arrow2` crate |
| JSON table | ✅ Supported | Uses `serde_json` |
| File server upload | ✅ Supported | HTTP/HTTPS via `reqwest` |
| File server download | ✅ Supported | HTTP/HTTPS via `reqwest` with retry |
| Size threshold | 500KB | Configurable |
| Async runtime | ✅ Supported | Uses `tokio` for async I/O |
| Type safety | ✅ Supported | Compile-time type checking via Rust enums |
### MicroPython
| Feature | Status | Notes |
@@ -818,6 +900,7 @@ function fileserver_download_handler(
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | Client-side rendering |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | Desktop Python |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | Desktop/Flutter/Web |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | Uses tokio + serde + arrow2 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | Memory-constrained |
### Browser Implementation Notes
@@ -832,16 +915,16 @@ The browser implementation ([`src/natsbridge_csr.js`](../src/natsbridge_csr.js))
### Payload Type Availability by Platform
| Payload Type | Julia | Node.js | Browser | Python | Dart | MicroPython |
|--------------|-------|---------|---------|--------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| Payload Type | Julia | Node.js | Browser | Python | Dart | Rust | MicroPython |
|--------------|-------|---------|---------|--------|------|------|-------------|
| `text` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `dictionary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `arrowtable` | ✅ | ✅ | ❌ | ✅ | ✅ | ✅ | ❌ |
| `jsontable` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ⚠️ |
| `image` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `audio` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `video` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| `binary` | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
---
@@ -979,6 +1062,13 @@ flowchart TD
| Dart | http | Latest | HTTP file server |
| Dart | uuid | Latest | UUID generation |
| Dart | dart-arrow | Latest | Arrow IPC support (Desktop/Flutter) |
| Rust | nats | Latest | NATS client |
| Rust | serde | Latest | JSON serialization |
| Rust | serde_json | Latest | JSON handling |
| Rust | tokio | Latest | Async runtime |
| Rust | reqwest | Latest | HTTP file server |
| Rust | uuid | Latest | UUID generation |
| Rust | arrow2 | Latest | Arrow IPC support |
| MicroPython | builtin | N/A | Limited implementation |
### Optional Dependencies
@@ -1031,6 +1121,8 @@ flowchart TD
| [`src/natsbridge_ssr.js`](../src/natsbridge_ssr.js) | Node.js | Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | FR-001 through FR-014, NFR-101 through NFR-405 |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | FR-005, FR-006, FR-012 |
### 20.3 External Dependencies
@@ -1049,6 +1141,17 @@ flowchart TD
| Python | nats-py | Latest | NATS client | FR-013, FR-014 |
| Python | aiohttp | Latest | HTTP file server | FR-008, FR-009 |
| Python | pyarrow | Latest | Arrow IPC support | FR-002, FR-012 |
| Dart | nats | Latest | NATS client | FR-013, FR-014 |
| Dart | http | Latest | HTTP file server | FR-008, FR-009 |
| Dart | uuid | Latest | UUID generation | FR-011, NFR-401 |
| Dart | dart-arrow | Latest | Arrow IPC support | FR-002, FR-012 |
| Rust | nats | Latest | NATS client | FR-013, FR-014 |
| Rust | serde | Latest | JSON serialization | FR-012, NFR-101, NFR-102 |
| Rust | serde_json | Latest | JSON handling | FR-012, NFR-101, NFR-102 |
| Rust | tokio | Latest | Async runtime | FR-013, FR-014 |
| Rust | reqwest | Latest | HTTP file server | FR-008, FR-009 |
| Rust | uuid | Latest | UUID generation | FR-011, NFR-401 |
| Rust | arrow2 | Latest | Arrow IPC support | FR-002, FR-012 |
| MicroPython | builtin | N/A | Limited implementation | FR-005, FR-006 |
---
@@ -1057,6 +1160,12 @@ 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.0.0
**Date**: 2026-03-23
**Version**: 1.2.0
**Date**: 2026-05-13
**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**, and **MicroPython** applications using NATS as the message bus.
This document provides the **end-to-end trace** for NATSBridge - the cross-platform bi-directional data bridge that enables seamless communication between **Julia**, **JavaScript**, **Python**, **Dart**, **Rust**, and **MicroPython** applications using NATS as the message bus.
This walkthrough serves as the primary onboarding guide for new developers and explains:
- **User scenarios** - Real-world use cases from developer perspective
@@ -213,23 +213,25 @@ 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
#### Step 5: Publish to NATS (Caller's Responsibility)
```javascript
await NATSBridge.NATSClient.connect("ws://localhost:4222");
await NATSBridge.NATSClient.publish("/agent/wine/api/v1/prompt", msgJson);
// 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);
```
**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
msg = NATS.subscription.next() # Get message from NATS
env = smartreceive(msg)
nats_msg = NATS.subscription.next() # Get message from NATS
env = smartreceive(String(nats_msg.payload))
# env["payloads"] is now:
# [
@@ -355,8 +357,8 @@ const response = await plikOneshotUpload(
```julia
# Julia backend
msg = NATS.subscription.next()
env = smartreceive(msg)
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
# NATSBridge automatically:
# 1. Extracts URL from payload
@@ -428,8 +430,8 @@ arrow_bytes = buf.getvalue()
```julia
# Julia backend
msg = NATS.subscription.next()
env = smartreceive(msg)
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
# env["payloads"][1] is now:
# ("data", DataFrame with id, name, score columns, "arrowtable")
@@ -461,7 +463,155 @@ env, msg_json = smartsend(
---
## User Scenario 4: MicroPython Device
## User Scenario 4: Rust Service with Type-Safe API
### Scenario Description
A Rust service needs to process messages from a Julia analytics pipeline and send typed results back. The Rust implementation leverages compile-time type safety via Rust enums and serde for serialization.
### Step-by-Step Flow
#### Step 1: Rust Service Receives Message
```rust
// Rust service - using tokio async runtime
use natsbridge::{smartreceive, MsgEnvelopeV1};
#[tokio::main]
async fn main() {
let conn = nats::connect("nats://localhost:4222").unwrap();
// Subscribe and receive messages
let mut sub = conn.subscribe("/agent/wine/api/v1/analyze").unwrap();
for msg in sub.messages() {
let envelope: MsgEnvelopeV1 = smartreceive(
&String::from_utf8_lossy(&msg.payload),
&Default::default(),
).await.unwrap();
// Type-safe payload access
for payload in &envelope.payloads {
match &payload.data {
Payload::ArrowTable(arrow_bytes) => {
// Process Arrow IPC data using arrow2
let table = arrow2::io::ipc::read::Reader::new(
std::io::Cursor::new(arrow_bytes.clone()),
);
println!("Received {} rows", table.len());
},
Payload::Text(text) => {
println!("Message: {}", text);
},
_ => println!("Received {} bytes of {} data",
match &payload.data {
Payload::Binary(b) => b.len(),
_ => 0,
},
payload.payload_type),
}
}
}
}
```
**Rationale**:
- **Type-safe payloads**: Rust enum discriminates between payload types at compile time
- **serde serialization**: Automatic JSON deserialization to `MsgEnvelopeV1`
- **tokio runtime**: Efficient async I/O for NATS and HTTP operations
- **arrow2 integration**: Direct Arrow IPC deserialization without intermediate format
#### Step 2: Rust Service Sends Processed Results
```rust
// Rust service sends results back with mixed payload types
use natsbridge::{smartsend, Payload, SmartsendOptions};
let results_df = /* processed Arrow table */;
let result_bytes = /* serialize to Arrow IPC */;
let (envelope, json_str) = smartsend(
"/agent/wine/api/v1/results",
&[
(
"results".to_string(),
Payload::ArrowTable(result_bytes),
"arrowtable".to_string(),
),
(
"summary".to_string(),
Payload::Text("Analysis complete: 1500 rows processed".to_string()),
"text".to_string(),
),
],
&SmartsendOptions {
broker_url: "nats://localhost:4222".to_string(),
reply_to: "/python/worker/v1/results".to_string(),
msg_purpose: "chat".to_string(),
..Default::default()
},
).await?;
// Caller publishes to NATS
conn.publish("/agent/wine/api/v1/results", &json_str)?;
```
**Rationale**:
- **Builder pattern**: `SmartsendOptions` provides clean configuration
- **Enum-based payloads**: Type safety prevents sending incorrect data types
- **Default options**: sensible defaults reduce boilerplate
- **Result<T, E>**: idiomatic Rust error handling
#### Step 3: Python/Julia Receives Rust Response
```python
# Python backend receives Rust response
env = await smartreceive(str(nats_msg.payload))
# env["payloads"][0] is now:
# ("results", arrow_table_data, "arrowtable")
# env["payloads"][1] is now:
# ("summary", "Analysis complete: 1500 rows processed", "text")
```
**Rationale**:
- **Cross-platform parity**: Rust envelope matches other platform envelopes exactly
- **Same JSON wire format**: No protocol translation needed
- **Type preservation**: Arrow IPC and text types preserved across all platforms
#### Step 4: Large File Transfer from Rust
```rust
// Rust service sends large binary file via link transport
let large_file_data: Vec<u8> = std::fs::read("/data/large_dataset.parquet")?;
let (envelope, json_str) = smartsend(
"/agent/wine/api/v1/upload",
&[
(
"dataset".to_string(),
Payload::Binary(large_file_data),
"binary".to_string(),
),
],
&SmartsendOptions {
broker_url: "nats://localhost:4222".to_string(),
fileserver_url: "http://localhost:8080".to_string(),
size_threshold: 500_000, // 0.5MB triggers link transport
..Default::default()
},
).await?;
```
**Rationale**:
- **Automatic transport selection**: Same 0.5MB threshold as other desktop platforms
- **reqwest integration**: Efficient HTTP client for file server upload/download
- **Exponential backoff**: Built-in retry with configurable parameters
- **Zero-copy where possible**: `Vec<u8>` passed directly without intermediate copies
---
## User Scenario 5: MicroPython Device
### Scenario Description
@@ -512,8 +662,8 @@ payload_b64 = base64.b64encode(json_bytes).decode('ascii')
```python
# Python backend
msg = await nats_consumer.next()
env = await smartreceive(msg)
nats_msg = await nats_consumer.next()
env = await smartreceive(str(nats_msg.payload))
# env["payloads"][0] is now:
# ("data", {"temperature": 25.5, "humidity": 60.0, ...}, "dictionary")
@@ -526,7 +676,7 @@ env = await smartreceive(msg)
---
## User Scenario 5: Cross-Platform Chat with Mixed Payloads
## User Scenario 6: Cross-Platform Chat with Mixed Payloads
### Scenario Description
@@ -561,8 +711,8 @@ const [env, msgJson] = await NATSBridge.smartsend(
```python
# Python (Backend)
msg = await nats_consumer.next()
env = await smartreceive(msg)
nats_msg = await nats_consumer.next()
env = await smartreceive(str(nats_msg.payload))
# env["payloads"] is now:
# [
@@ -580,8 +730,8 @@ env = await smartreceive(msg)
```julia
# Julia (Backend)
msg = NATS.subscription.next()
env = smartreceive(msg)
nats_msg = NATS.subscription.next()
env = smartreceive(String(nats_msg.payload))
# env["payloads"] is now:
# [
@@ -726,7 +876,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` | `1000000` | Size threshold in bytes |
| `SIZE_THRESHOLD` | `500000` | Size threshold in bytes (0.5MB) |
---
@@ -761,6 +911,7 @@ log_trace(correlation_id, "Published to NATS")
| [`src/natsbridge_csr.js`](../src/natsbridge_csr.js) | Browser | JSON table only, WebSocket NATS | specification.md:2-19 (all sections) |
| [`src/natsbridge.py`](../src/natsbridge.py) | Python | Arrow IPC, async/await | specification.md:2-19 (all sections) |
| [`src/natsbridge.dart`](../src/natsbridge.dart) | Dart | Full feature set, Arrow IPC, async/await | specification.md:2-19 (all sections) |
| [`src/natsbridge.rs`](../src/natsbridge.rs) | Rust | Full feature set, Arrow IPC, async/await, type-safe | specification.md:2-19 (all sections) |
| [`src/natsbridge_mpy.py`](../src/natsbridge_mpy.py) | MicroPython | Limited to direct transport | specification.md:2-19 (all sections) |
---
@@ -769,6 +920,13 @@ log_trace(correlation_id, "Published to NATS")
| Date | Version | Changes | Specification Reference |
|------|---------|---------|------------------------|
| 2026-05-13 | 1.3.0 | Added Rust support with tokio, serde, and arrow2 | All sections |
| - | - | Added Rust user scenario (User Scenario 4) | specification.md:11 (Rust API) |
| - | - | Updated scenario numbering (MicroPython → Scenario 5, Cross-Platform → Scenario 6) | All sections |
| 2026-05-13 | 1.2.0 | Aligned with ground truth implementation (src/NATSBridge.jl) | All sections |
| - | - | Updated smartreceive calls to use String(nats_msg.payload) pattern | All sections |
| - | - | Removed NATSClient.publish() calls (caller responsible for NATS publishing) | All sections |
| - | - | 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
View File

@@ -1,56 +0,0 @@
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

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

View File

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

View File

@@ -43,7 +43,7 @@
module NATSBridge
using NATS, JSON, Arrow, HTTP, UUIDs, Dates, Base64, PrettyPrinting, DataFrames
using JSON, Arrow, HTTP, UUIDs, Dates, Base64, PrettyPrinting, DataFrames
# ---------------------------------------------- 100 --------------------------------------------- #
# Constants
@@ -340,25 +340,29 @@ 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 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.
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.
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. Converts envelope to JSON string and optionally publishes to NATS
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)
# Arguments:
- `subject::String` - NATS subject to publish the message to
- `data::AbstractArray{Tuple{String, Any, String}}` - List of (dataname, data, type) tuples to send
- `data::AbstractArray{Tuple{String, T1, String}, 1}` - List of (dataname, data, type) tuples to send
- `dataname::String` - Name of the payload
- `data::Any` - The actual data to send
- `data::T1` - The actual data to send (any type supported by `_serialize_data`)
- `payload_type::String` - Payload type: "text", "dictionary", "arrowtable", "jsontable", "image", "audio", "video", "binary"
- No standalone `type` parameter - type is specified per payload
@@ -367,17 +371,15 @@ Each payload can have a different type, enabling mixed-content messages (e.g., c
- `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
- `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)
- `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)
# Return:
- `::Tuple{msg_envelope_v1, String}` - A tuple containing:
@@ -412,8 +414,9 @@ env, msg_json = smartsend("chat.subject", [
("audio_clip", audio_data, "audio")
])
# Publish the JSON string directly using NATS request-reply pattern
# reply = NATS.request(broker_url, subject, env_json_str; reply_to=reply_to_topic)
# Publish the JSON string directly using NATS (manual publish)
# conn = NATS.connect(broker_url)
# NATS.publish(conn, subject, env_json_str)
```
"""
function smartsend(
@@ -438,8 +441,6 @@ 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}
@@ -539,13 +540,15 @@ 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(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
# 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
return (env, env_json_str)
end
@@ -700,73 +703,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
@@ -783,7 +786,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::NATS.Msg` - NATS message to process
- `msg_json_str::String` - JSON string from NATS message payload (e.g., `String(nats_msg.payload)`)
# Keyword Arguments:
- `fileserver_download_handler::Function = _fetch_with_backoff` - Function to handle downloading data from file server URLs
@@ -798,19 +801,21 @@ A HTTP file server is required along with its download function.
```jldoctest
# Receive and process message
msg = nats_message # NATS message
env = smartreceive(msg; fileserver_download_handler=_fetch_with_backoff, max_retries=5, base_delay=100, max_delay=5000)
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["payloads"] = [("dataname1", data1, "type1"), ("dataname2", data2, "type2"), ...]
```
"""
function smartreceive(
msg::NATS.Msg;
msg_json_str::String; # get it from String(nats_msg.payload)
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(String(msg.payload))
env_json_obj = JSON.parse(msg_json_str)
log_trace(env_json_obj["correlation_id"], "Processing received message") # Log message processing start
# Process all payloads in the envelope

View File

@@ -1,782 +0,0 @@
/// 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;

1230
src/natsbridge.rs Normal file

File diff suppressed because it is too large Load Diff

View File

@@ -1,74 +0,0 @@
/// 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

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