diff --git a/examples/micropython_example.py b/examples/micropython_example.py deleted file mode 100644 index 59db61f..0000000 --- a/examples/micropython_example.py +++ /dev/null @@ -1,221 +0,0 @@ -""" -Micropython NATS Bridge - Simple Example - -This example demonstrates the basic usage of the NATSBridge for Micropython. -""" - -import sys -sys.path.insert(0, "../src") - -from nats_bridge import smartsend, smartreceive, log_trace -import json - - -def example_simple_chat(): - """ - Simple chat example: Send text messages via NATS. - - Sender (this script): - - Sends a text message to NATS - - Uses direct transport (no fileserver needed) - - Receiver (separate script): - - Listens to NATS - - Receives and processes the message - """ - print("=== Simple Chat Example ===") - print() - - # Define the message data as list of (dataname, data, type) tuples - data = [ - ("message", "Hello from Micropython!", "text") - ] - - # Send the message - env = smartsend( - "/chat/room1", - data, - nats_url="nats://localhost:4222", - msg_purpose="chat", - sender_name="micropython-client" - ) - - print("Message sent!") - print(" Subject: {}".format(env.send_to)) - print(" Correlation ID: {}".format(env.correlation_id)) - print(" Payloads: {}".format(len(env.payloads))) - print() - - # Expected receiver output: - print("Expected receiver output:") - print(" [timestamp] [Correlation: ...] Starting smartsend for subject: /chat/room1") - print(" [timestamp] [Correlation: ...] Serialized payload 'message' (type: text) size: 22 bytes") - print(" [timestamp] [Correlation: ...] Using direct transport for 22 bytes") - print(" [timestamp] [Correlation: ...] Message published to /chat/room1") - print() - - return env - - -def example_send_json(): - """ - Example: Send JSON configuration to a Micropython device. - - This demonstrates sending structured data (dictionary type). - """ - print("\n=== Send JSON Configuration ===") - print() - - # Define configuration as dictionary - config = { - "wifi_ssid": "MyNetwork", - "wifi_password": "password123", - "server_host": "mqtt.example.com", - "server_port": 1883, - "update_interval": 60 - } - - # Send configuration - data = [ - ("device_config", config, "dictionary") - ] - - env = smartsend( - "/device/config", - data, - nats_url="nats://localhost:4222", - msg_purpose="updateStatus", - sender_name="server" - ) - - print("Configuration sent!") - print(" Subject: {}".format(env.send_to)) - print(" Payloads: {}".format(len(env.payloads))) - print() - - return env - - -def example_receive_message(msg): - """ - Example: Receive and process a NATS message. - - Args: - msg: The NATS message received (should be dict or JSON string) - - Returns: - list: List of (dataname, data, type) tuples - """ - print("\n=== Receive Message ===") - print() - - # Process the message - payloads = smartreceive( - msg, - fileserver_download_handler=None, # Not needed for direct transport - max_retries=3, - base_delay=100, - max_delay=1000 - ) - - print("Received {} payload(s):".format(len(payloads))) - for dataname, data, type in payloads: - print(" - {}: {} (type: {})".format(dataname, data, type)) - - return payloads - - -def example_mixed_content(): - """ - Example: Send mixed content (text + dictionary + binary). - - This demonstrates the multi-payload capability. - """ - print("\n=== Mixed Content Example ===") - print() - - # Create mixed content - image_data = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR" # Example PNG header - - data = [ - ("message_text", "Hello with image!", "text"), - ("user_config", {"theme": "dark", "notifications": True}, "dictionary"), - ("user_avatar", image_data, "binary") - ] - - env = smartsend( - "/chat/mixed", - data, - nats_url="nats://localhost:4222", - msg_purpose="chat", - sender_name="micropython-client" - ) - - print("Mixed content sent!") - print(" Subject: {}".format(env.send_to)) - print(" Payloads:") - for p in env.payloads: - print(" - {} (transport: {}, type: {}, size: {} bytes)".format( - p.dataname, p.transport, p.type, p.size)) - - return env - - -def example_reply(): - """ - Example: Send a message with reply-to functionality. - - This demonstrates request-response pattern. - """ - print("\n=== Request-Response Example ===") - print() - - # Send command - data = [ - ("command", {"action": "read_sensor", "sensor_id": "temp1"}, "dictionary") - ] - - env = smartsend( - "/device/command", - data, - nats_url="nats://localhost:4222", - msg_purpose="command", - sender_name="server", - reply_to="/device/response", - reply_to_msg_id="cmd-001" - ) - - print("Command sent!") - print(" Subject: {}".format(env.send_to)) - print(" Reply To: {}".format(env.reply_to)) - print(" Reply To Msg ID: {}".format(env.reply_to_msg_id)) - print() - - print("Expected receiver behavior:") - print(" 1. Receive command on /device/command") - print(" 2. Process command") - print(" 3. Send response to /device/response") - print(" 4. Include replyToMsgId in response") - - return env - - -if __name__ == "__main__": - print("Micropython NATS Bridge Examples") - print("================================") - print() - - # Run examples - example_simple_chat() - example_send_json() - example_mixed_content() - example_reply() - - print("\n=== Examples Completed ===") - print() - print("To run these examples, you need:") - print(" 1. A running NATS server at nats://localhost:4222") - print(" 2. Import the nats_bridge module") - print(" 3. Call the desired example function") - print() - print("For more examples, see test/test_micropython_basic.py") \ No newline at end of file diff --git a/examples/tutorial.md b/examples/tutorial.md new file mode 100644 index 0000000..fcc71dc --- /dev/null +++ b/examples/tutorial.md @@ -0,0 +1,604 @@ +# NATSBridge Tutorial + +A step-by-step guide to get started with NATSBridge - a high-performance, bi-directional data bridge for **Julia**, **JavaScript**, and **Python/Micropython**. + +## Table of Contents + +1. [Overview](#overview) +2. [Prerequisites](#prerequisites) +3. [Installation](#installation) +4. [Quick Start](#quick-start) +5. [Basic Examples](#basic-examples) +6. [Advanced Usage](#advanced-usage) +7. [Cross-Platform Communication](#cross-platform-communication) + +--- + +## Overview + +NATSBridge enables seamless communication between Julia, JavaScript, and Python/Micropython applications through NATS, with automatic transport selection based on payload size: + +- **Direct Transport**: Payloads < 1MB are sent directly via NATS (Base64 encoded) +- **Link Transport**: Payloads >= 1MB are uploaded to an HTTP file server and referenced via URL + +### Supported Payload Types + +| Type | Description | +|------|-------------| +| `text` | Plain text strings | +| `dictionary` | JSON-serializable dictionaries | +| `table` | Tabular data (Arrow IPC format) | +| `image` | Image data (PNG, JPG bytes) | +| `audio` | Audio data (WAV, MP3 bytes) | +| `video` | Video data (MP4, AVI bytes) | +| `binary` | Generic binary data | + +--- + +## Prerequisites + +Before you begin, ensure you have: + +1. **NATS Server** running (or accessible) +2. **HTTP File Server** (optional, for large payloads > 1MB) +3. **One of the supported platforms**: Julia, JavaScript (Node.js), or Python/Micropython + +--- + +## Installation + +### Julia + +```julia +using Pkg +Pkg.add("NATS") +Pkg.add("Arrow") +Pkg.add("JSON3") +Pkg.add("HTTP") +Pkg.add("UUIDs") +Pkg.add("Dates") +``` + +### JavaScript + +```bash +npm install nats.js apache-arrow uuid base64-url +``` + +### Python/Micropython + +1. Copy `src/nats_bridge.py` to your device +2. Install dependencies: + +**For Python (desktop):** +```bash +pip install nats-py +``` + +**For Micropython:** +- `urequests` for HTTP requests +- `base64` for base64 encoding (built-in) +- `json` for JSON handling (built-in) + +--- + +## Quick Start + +### Step 1: Start NATS Server + +```bash +docker run -p 4222:4222 nats:latest +``` + +### Step 2: Start HTTP File Server (Optional) + +```bash +# Create a directory for file uploads +mkdir -p /tmp/fileserver + +# Use Python's built-in server +python3 -m http.server 8080 --directory /tmp/fileserver +``` + +### Step 3: Send Your First Message + +#### Python/Micropython + +```python +from nats_bridge import smartsend + +# Send a text message +data = [("message", "Hello World", "text")] +env = smartsend("/chat/room1", data, nats_url="nats://localhost:4222") +print("Message sent!") +``` + +#### JavaScript + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +// Send a text message +await smartsend("/chat/room1", [ + { dataname: "message", data: "Hello World", type: "text" } +], { natsUrl: "nats://localhost:4222" }); + +console.log("Message sent!"); +``` + +#### Julia + +```julia +using NATSBridge + +# Send a text message +data = [("message", "Hello World", "text")] +env = smartsend("/chat/room1", data, nats_url="nats://localhost:4222") +println("Message sent!") +``` + +### Step 4: Receive Messages + +#### Python/Micropython + +```python +from nats_bridge import smartreceive + +# Receive and process message +envelope = smartreceive(msg) +for dataname, data, type in envelope["payloads"]: + print(f"Received {dataname}: {data}") +``` + +#### JavaScript + +```javascript +const { smartreceive } = require('./src/NATSBridge'); + +// Receive and process message +const envelope = await smartreceive(msg); +for (const payload of envelope.payloads) { + console.log(`Received ${payload.dataname}: ${payload.data}`); +} +``` + +#### Julia + +```julia +using NATSBridge + +# Receive and process message +envelope = smartreceive(msg, fileserverDownloadHandler) +for (dataname, data, type) in envelope["payloads"] + println("Received $dataname: $data") +end +``` + +--- + +## Basic Examples + +### Example 1: Sending a Dictionary + +#### Python/Micropython + +```python +from nats_bridge import smartsend + +# Create configuration dictionary +config = { + "wifi_ssid": "MyNetwork", + "wifi_password": "password123", + "update_interval": 60 +} + +# Send as dictionary type +data = [("config", config, "dictionary")] +env = smartsend("/device/config", data, nats_url="nats://localhost:4222") +``` + +#### JavaScript + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +const config = { + wifi_ssid: "MyNetwork", + wifi_password: "password123", + update_interval: 60 +}; + +await smartsend("/device/config", [ + { dataname: "config", data: config, type: "dictionary" } +]); +``` + +#### Julia + +```julia +using NATSBridge + +config = Dict( + "wifi_ssid" => "MyNetwork", + "wifi_password" => "password123", + "update_interval" => 60 +) + +data = [("config", config, "dictionary")] +smartsend("/device/config", data) +``` + +### Example 2: Sending Binary Data (Image) + +#### Python/Micropython + +```python +from nats_bridge import smartsend + +# Read image file +with open("image.png", "rb") as f: + image_data = f.read() + +# Send as binary type +data = [("user_image", image_data, "binary")] +env = smartsend("/chat/image", data, nats_url="nats://localhost:4222") +``` + +#### JavaScript + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +// Read image file (Node.js) +const fs = require('fs'); +const image_data = fs.readFileSync('image.png'); + +await smartsend("/chat/image", [ + { dataname: "user_image", data: image_data, type: "binary" } +]); +``` + +#### Julia + +```julia +using NATSBridge + +# Read image file +image_data = read("image.png") + +data = [("user_image", image_data, "binary")] +smartsend("/chat/image", data) +``` + +### Example 3: Request-Response Pattern + +#### Python/Micropython (Requester) + +```python +from nats_bridge import smartsend + +# Send command with reply-to +data = [("command", {"action": "read_sensor"}, "dictionary")] +env = smartsend( + "/device/command", + data, + nats_url="nats://localhost:4222", + reply_to="/device/response", + reply_to_msg_id="cmd-001" +) +``` + +#### JavaScript (Responder) + +```javascript +const { smartreceive, smartsend } = require('./src/NATSBridge'); + +// Subscribe to command topic +const sub = nc.subscribe("/device/command"); + +for await (const msg of sub) { + const envelope = await smartreceive(msg); + + // Process command + for (const payload of envelope.payloads) { + if (payload.dataname === "command") { + const command = payload.data; + + if (command.action === "read_sensor") { + // Read sensor and send response + const response = { + sensor_id: "sensor-001", + value: 42.5, + timestamp: new Date().toISOString() + }; + + await smartsend("/device/response", [ + { dataname: "sensor_data", data: response, type: "dictionary" } + ], { + reply_to: envelope.replyTo, + reply_to_msg_id: envelope.msgId + }); + } + } + } +} +``` + +--- + +## Advanced Usage + +### Example 4: Large Payloads (File Server) + +For payloads larger than 1MB, NATSBridge automatically uses the file server: + +#### Python/Micropython + +```python +from nats_bridge import smartsend +import os + +# Create large data (> 1MB) +large_data = os.urandom(2_000_000) # 2MB of random data + +# Send with file server URL +env = smartsend( + "/data/large", + [("large_file", large_data, "binary")], + nats_url="nats://localhost:4222", + fileserver_url="http://localhost:8080", + size_threshold=1_000_000 +) + +# The envelope will contain the download URL +print(f"File uploaded to: {env.payloads[0].data}") +``` + +#### JavaScript + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +// Create large data (> 1MB) +const largeData = new ArrayBuffer(2_000_000); +const view = new Uint8Array(largeData); +view.fill(42); // Fill with some data + +await smartsend("/data/large", [ + { dataname: "large_file", data: largeData, type: "binary" } +], { + fileserverUrl: "http://localhost:8080", + sizeThreshold: 1_000_000 +}); +``` + +#### Julia + +```julia +using NATSBridge + +# Create large data (> 1MB) +large_data = rand(UInt8, 2_000_000) + +env = smartsend( + "/data/large", + [("large_file", large_data, "binary")], + fileserver_url="http://localhost:8080" +) + +# The envelope will contain the download URL +println("File uploaded to: $(env.payloads[1].data)") +``` + +### Example 5: Mixed Content (Chat with Text + Image) + +NATSBridge supports sending multiple payloads with different types in a single message: + +#### Python/Micropython + +```python +from nats_bridge import smartsend + +# Read image file +with open("avatar.png", "rb") as f: + image_data = f.read() + +# Send mixed content +data = [ + ("message_text", "Hello with image!", "text"), + ("user_avatar", image_data, "image") +] + +env = smartsend("/chat/mixed", data, nats_url="nats://localhost:4222") +``` + +#### JavaScript + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +const fs = require('fs'); + +await smartsend("/chat/mixed", [ + { + dataname: "message_text", + data: "Hello with image!", + type: "text" + }, + { + dataname: "user_avatar", + data: fs.readFileSync("avatar.png"), + type: "image" + } +]); +``` + +#### Julia + +```julia +using NATSBridge + +image_data = read("avatar.png") + +data = [ + ("message_text", "Hello with image!", "text"), + ("user_avatar", image_data, "image") +] + +smartsend("/chat/mixed", data) +``` + +### Example 6: Table Data (Arrow IPC) + +For tabular data, NATSBridge uses Apache Arrow IPC format: + +#### Python/Micropython + +```python +from nats_bridge import smartsend +import pandas as pd + +# Create DataFrame +df = pd.DataFrame({ + "id": [1, 2, 3], + "name": ["Alice", "Bob", "Charlie"], + "score": [95, 88, 92] +}) + +# Send as table type +data = [("students", df, "table")] +env = smartsend("/data/students", data, nats_url="nats://localhost:4222") +``` + +#### Julia + +```julia +using NATSBridge +using DataFrames + +# Create DataFrame +df = DataFrame( + id = [1, 2, 3], + name = ["Alice", "Bob", "Charlie"], + score = [95, 88, 92] +) + +data = [("students", df, "table")] +smartsend("/data/students", data) +``` + +--- + +## Cross-Platform Communication + +NATSBridge enables seamless communication between different platforms: + +### Julia ↔ JavaScript + +#### Julia Sender + +```julia +using NATSBridge + +# Send dictionary from Julia to JavaScript +config = Dict("step_size" => 0.01, "iterations" => 1000) +data = [("config", config, "dictionary")] +smartsend("/analysis/config", data, nats_url="nats://localhost:4222") +``` + +#### JavaScript Receiver + +```javascript +const { smartreceive } = require('./src/NATSBridge'); + +// Receive dictionary from Julia +const envelope = await smartreceive(msg); +for (const payload of envelope.payloads) { + if (payload.type === "dictionary") { + console.log("Received config:", payload.data); + // payload.data = { step_size: 0.01, iterations: 1000 } + } +} +``` + +### JavaScript ↔ Python + +#### JavaScript Sender + +```javascript +const { smartsend } = require('./src/NATSBridge'); + +await smartsend("/data/transfer", [ + { dataname: "message", data: "Hello from JS!", type: "text" } +]); +``` + +#### Python Receiver + +```python +from nats_bridge import smartreceive + +envelope = smartreceive(msg) +for dataname, data, type in envelope["payloads"]: + if type == "text": + print(f"Received from JS: {data}") +``` + +### Python ↔ Julia + +#### Python Sender + +```python +from nats_bridge import smartsend + +data = [("message", "Hello from Python!", "text")] +smartsend("/chat/python", data) +``` + +#### Julia Receiver + +```julia +using NATSBridge + +envelope = smartreceive(msg, fileserverDownloadHandler) +for (dataname, data, type) in envelope["payloads"] + if type == "text" + println("Received from Python: $data") + end +end +``` + +--- + +## Next Steps + +1. **Explore the test directory** for more examples +2. **Check the documentation** for advanced configuration options +3. **Join the community** to share your use cases + +--- + +## Troubleshooting + +### Connection Issues + +- Ensure NATS server is running: `docker ps | grep nats` +- Check firewall settings +- Verify NATS URL configuration + +### File Server Issues + +- Ensure file server is running and accessible +- Check upload permissions +- Verify file server URL configuration + +### Serialization Errors + +- Verify data type matches the specified type +- Check that binary data is in the correct format (bytes/Vector{UInt8}) + +--- + +## License + +MIT \ No newline at end of file diff --git a/examples/tutorial_julia.md b/examples/tutorial_julia.md deleted file mode 100644 index 2e395df..0000000 --- a/examples/tutorial_julia.md +++ /dev/null @@ -1,634 +0,0 @@ -# NATSBridge.jl Tutorial - -A comprehensive tutorial for learning how to use NATSBridge.jl for bi-directional communication between Julia and JavaScript services using NATS. - -## Table of Contents - -1. [What is NATSBridge.jl?](#what-is-natsbridgejl) -2. [Key Concepts](#key-concepts) -3. [Installation](#installation) -4. [Basic Usage](#basic-usage) -5. [Payload Types](#payload-types) -6. [Transport Strategies](#transport-strategies) -7. [Advanced Features](#advanced-features) -8. [Complete Examples](#complete-examples) - ---- - -## What is NATSBridge.jl? - -NATSBridge.jl is a Julia module that provides a high-level API for sending and receiving data across network boundaries using NATS as the message bus. It implements the **Claim-Check pattern** for handling large payloads efficiently. - -### Core Features - -- **Bi-directional communication**: Julia ↔ JavaScript -- **Smart transport selection**: Automatic direct vs link transport based on payload size -- **Multi-payload support**: Send multiple payloads of different types in a single message -- **Claim-check pattern**: Upload large files to HTTP server, send only URLs via NATS -- **Type-aware serialization**: Different serialization strategies for different data types - ---- - -## Key Concepts - -### 1. msgEnvelope_v1 (Message Envelope) - -The `msgEnvelope_v1` structure provides a comprehensive message format for bidirectional communication: - -```julia -struct msgEnvelope_v1 - correlationId::String # Unique identifier to track messages - msgId::String # This message id - timestamp::String # Message published timestamp - - sendTo::String # Topic/subject the sender sends to - msgPurpose::String # Purpose (ACK | NACK | updateStatus | shutdown | chat) - senderName::String # Sender name (e.g., "agent-wine-web-frontend") - senderId::String # Sender id (uuid4) - receiverName::String # Message receiver name (e.g., "agent-backend") - receiverId::String # Message receiver id (uuid4 or nothing for broadcast) - replyTo::String # Topic to reply to - replyToMsgId::String # Message id this message is replying to - brokerURL::String # NATS server address - - metadata::Dict{String, Any} - payloads::AbstractArray{msgPayload_v1} # Multiple payloads stored here -end -``` - -### 2. msgPayload_v1 (Payload Structure) - -The `msgPayload_v1` structure provides flexible payload handling: - -```julia -struct msgPayload_v1 - id::String # Id of this payload (e.g., "uuid4") - dataname::String # Name of this payload (e.g., "login_image") - type::String # "text | dictionary | table | image | audio | video | binary" - transport::String # "direct | link" - encoding::String # "none | json | base64 | arrow-ipc" - size::Integer # Data size in bytes - data::Any # Payload data in case of direct transport or a URL in case of link - metadata::Dict{String, Any} # Dict("checksum" => "sha256_hash", ...) -end -``` - -### 3. Standard API Format - -The system uses a **standardized list-of-tuples format** for all payload operations: - -```julia -# Input format for smartsend (always a list of tuples with type info) -[(dataname1, data1, type1), (dataname2, data2, type2), ...] - -# Output format for smartreceive (always returns a list of tuples) -[(dataname1, data1, type1), (dataname2, data2, type2), ...] -``` - -**Important**: Even when sending a single payload, you must wrap it in a list. - ---- - -## Installation - -```julia -using Pkg -Pkg.add("NATS") -Pkg.add("JSON") -Pkg.add("Arrow") -Pkg.add("HTTP") -Pkg.add("UUIDs") -Pkg.add("Dates") -Pkg.add("Base64") -Pkg.add("PrettyPrinting") -Pkg.add("DataFrames") -``` - -Then include the NATSBridge module: - -```julia -include("NATSBridge.jl") -using .NATSBridge -``` - ---- - -## Basic Usage - -### Sending Data (smartsend) - -```julia -using NATSBridge - -# Send a simple dictionary -data = Dict("key" => "value") -env = NATSBridge.smartsend("my.subject", [("dataname1", data, "dictionary")]) -``` - -### Receiving Data (smartreceive) - -```julia -using NATSBridge - -# Subscribe to a NATS subject -NATS.subscribe("my.subject") do msg - # Process the message - result = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - # result is a list of (dataname, data, type) tuples - for (dataname, data, type) in result - println("Received $dataname of type $type") - println("Data: $data") - end -end -``` - ---- - -## Payload Types - -NATSBridge.jl supports the following payload types: - -| Type | Description | Serialization | -|------|-------------|---------------| -| `text` | Plain text | UTF-8 encoding | -| `dictionary` | JSON-serializable data (Dict, NamedTuple) | JSON | -| `table` | Tabular data (DataFrame, array of structs) | Apache Arrow IPC | -| `image` | Image data (Bitmap, PNG/JPG bytes) | Binary | -| `audio` | Audio data (WAV, MP3 bytes) | Binary | -| `video` | Video data (MP4, AVI bytes) | Binary | -| `binary` | Generic binary data | Binary | - ---- - -## Transport Strategies - -NATSBridge.jl automatically selects the appropriate transport strategy based on payload size: - -### Direct Transport (< 1MB) - -Small payloads are encoded as Base64 and sent directly over NATS. - -```julia -# Small data (< 1MB) - uses direct transport -small_data = rand(1000) # ~8KB -env = NATSBridge.smartsend("small", [("data", small_data, "table")]) -``` - -### Link Transport (≥ 1MB) - -Large payloads are uploaded to an HTTP file server, and only the URL is sent via NATS. - -```julia -# Large data (≥ 1MB) - uses link transport -large_data = rand(10_000_000) # ~80MB -env = NATSBridge.smartsend("large", [("data", large_data, "table")]) -``` - ---- - -## Complete Examples - -### Example 1: Text Message - -**Sender:** -```julia -using NATSBridge -using UUIDs - -const SUBJECT = "/NATSBridge_text_test" -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function test_text_send() - small_text = "Hello, this is a small text message." - large_text = join(["Line $i: " for i in 1:50000], "") - - data1 = ("small_text", small_text, "text") - data2 = ("large_text", large_text, "text") - - env = NATSBridge.smartsend( - SUBJECT, - [data1, data2], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "text_sender" - ) -end -``` - -**Receiver:** -```julia -using NATSBridge - -const SUBJECT = "/NATSBridge_text_test" -const NATS_URL = "nats://localhost:4222" - -function test_text_receive() - conn = NATS.connect(NATS_URL) - NATS.subscribe(conn, SUBJECT) do msg - result = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in result - if data_type == "text" - println("Received text: $data") - write("./received_$dataname.txt", data) - end - end - end - sleep(120) - NATS.drain(conn) -end -``` - -### Example 2: Dictionary (JSON) Message - -**Sender:** -```julia -using NATSBridge -using UUIDs - -const SUBJECT = "/NATSBridge_dict_test" -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function test_dict_send() - small_dict = Dict("name" => "Alice", "age" => 30) - large_dict = Dict("ids" => collect(1:50000), "names" => ["User_$i" for i in 1:50000]) - - data1 = ("small_dict", small_dict, "dictionary") - data2 = ("large_dict", large_dict, "dictionary") - - env = NATSBridge.smartsend( - SUBJECT, - [data1, data2], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat" - ) -end -``` - -**Receiver:** -```julia -using NATSBridge - -const SUBJECT = "/NATSBridge_dict_test" -const NATS_URL = "nats://localhost:4222" - -function test_dict_receive() - conn = NATS.connect(NATS_URL) - NATS.subscribe(conn, SUBJECT) do msg - result = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in result - if data_type == "dictionary" - println("Received dictionary: $data") - write("./received_$dataname.json", JSON.json(data, 2)) - end - end - end - sleep(120) - NATS.drain(conn) -end -``` - -### Example 3: DataFrame (Table) Message - -**Sender:** -```julia -using NATSBridge -using DataFrames -using UUIDs - -const SUBJECT = "/NATSBridge_table_test" -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function test_table_send() - small_df = DataFrame(id = 1:10, name = ["Alice", "Bob", "Charlie"], score = [95, 88, 92]) - large_df = DataFrame(id = 1:50000, name = ["User_$i" for i in 1:50000], score = rand(1:100, 50000)) - - data1 = ("small_table", small_df, "table") - data2 = ("large_table", large_df, "table") - - env = NATSBridge.smartsend( - SUBJECT, - [data1, data2], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat" - ) -end -``` - -**Receiver:** -```julia -using NATSBridge -using DataFrames - -const SUBJECT = "/NATSBridge_table_test" -const NATS_URL = "nats://localhost:4222" - -function test_table_receive() - conn = NATS.connect(NATS_URL) - NATS.subscribe(conn, SUBJECT) do msg - result = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in result - if data_type == "table" - data = DataFrame(data) - println("Received DataFrame with $(size(data, 1)) rows") - display(data[1:min(5, size(data, 1)), :]) - end - end - end - sleep(120) - NATS.drain(conn) -end -``` - -### Example 4: Mixed Content (Chat with Text, Image, Audio) - -**Sender:** -```julia -using NATSBridge -using DataFrames -using UUIDs - -const SUBJECT = "/NATSBridge_mix_test" -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function test_mix_send() - # Text data - text_data = "Hello! This is a test chat message. 🎉" - - # Dictionary data - dict_data = Dict("type" => "chat", "sender" => "serviceA") - - # Small table data - table_data_small = DataFrame(id = 1:10, name = ["msg_$i" for i in 1:10]) - - # Large table data (link transport) - table_data_large = DataFrame(id = 1:150_000, name = ["msg_$i" for i in 1:150_000]) - - # Small image data (direct transport) - image_data = UInt8[rand(1:255) for _ in 1:100] - - # Large image data (link transport) - large_image_data = UInt8[rand(1:255) for _ in 1:1_500_000] - - # Small audio data (direct transport) - audio_data = UInt8[rand(1:255) for _ in 1:100] - - # Large audio data (link transport) - large_audio_data = UInt8[rand(1:255) for _ in 1:1_500_000] - - # Small video data (direct transport) - video_data = UInt8[rand(1:255) for _ in 1:150] - - # Large video data (link transport) - large_video_data = UInt8[rand(1:255) for _ in 1:1_500_000] - - # Small binary data (direct transport) - binary_data = UInt8[rand(1:255) for _ in 1:200] - - # Large binary data (link transport) - large_binary_data = UInt8[rand(1:255) for _ in 1:1_500_000] - - # Create payloads list - mixed content - payloads = [ - # Small data (direct transport) - ("chat_text", text_data, "text"), - ("chat_json", dict_data, "dictionary"), - ("chat_table_small", table_data_small, "table"), - - # Large data (link transport) - ("chat_table_large", table_data_large, "table"), - ("user_image_large", large_image_data, "image"), - ("audio_clip_large", large_audio_data, "audio"), - ("video_clip_large", large_video_data, "video"), - ("binary_file_large", large_binary_data, "binary") - ] - - env = NATSBridge.smartsend( - SUBJECT, - payloads, - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "mix_sender" - ) -end -``` - -**Receiver:** -```julia -using NATSBridge -using DataFrames - -const SUBJECT = "/NATSBridge_mix_test" -const NATS_URL = "nats://localhost:4222" - -function test_mix_receive() - conn = NATS.connect(NATS_URL) - NATS.subscribe(conn, SUBJECT) do msg - result = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - println("Received $(length(result)) payloads") - - for (dataname, data, data_type) in result - println("\n=== Payload: $dataname (type: $data_type) ===") - - if data_type == "text" - println(" Type: String") - println(" Length: $(length(data)) characters") - - elseif data_type == "dictionary" - println(" Type: JSON Object") - println(" Keys: $(keys(data))") - - elseif data_type == "table" - data = DataFrame(data) - println(" Type: DataFrame") - println(" Dimensions: $(size(data, 1)) rows x $(size(data, 2)) columns") - - elseif data_type == "image" - println(" Type: Vector{UInt8}") - println(" Size: $(length(data)) bytes") - write("./received_$dataname.bin", data) - - elseif data_type == "audio" - println(" Type: Vector{UInt8}") - println(" Size: $(length(data)) bytes") - write("./received_$dataname.bin", data) - - elseif data_type == "video" - println(" Type: Vector{UInt8}") - println(" Size: $(length(data)) bytes") - write("./received_$dataname.bin", data) - - elseif data_type == "binary" - println(" Type: Vector{UInt8}") - println(" Size: $(length(data)) bytes") - write("./received_$dataname.bin", data) - end - end - end - sleep(120) - NATS.drain(conn) -end -``` - ---- - -## Best Practices - -1. **Always wrap payloads in a list** - Even for single payloads: `[("dataname", data, "type")]` -2. **Use appropriate transport** - Let NATSBridge handle size-based routing (default 1MB threshold) -3. **Customize size threshold** - Use `size_threshold` parameter to adjust the direct/link split -4. **Provide fileserver handler** - Implement `fileserverUploadHandler` for link transport -5. **Include correlation IDs** - Track messages across distributed systems -6. **Handle errors** - Implement proper error handling for network failures -7. **Close connections** - Ensure NATS connections are properly closed using `NATS.drain()` - ---- - -## Conclusion - -NATSBridge.jl provides a powerful abstraction for bi-directional communication between Julia and JavaScript services. By understanding the key concepts and following the best practices, you can build robust, scalable applications that leverage the full power of NATS messaging. - -For more information, see: -- [`docs/architecture.md`](./architecture.md) - Detailed architecture documentation -- [`docs/implementation.md`](./implementation.md) - Implementation details \ No newline at end of file diff --git a/examples/walkthrough.md b/examples/walkthrough.md new file mode 100644 index 0000000..6897238 --- /dev/null +++ b/examples/walkthrough.md @@ -0,0 +1,1045 @@ +# NATSBridge Walkthrough + +A comprehensive guide to building real-world applications with NATSBridge. + +## Table of Contents + +1. [Introduction](#introduction) +2. [Architecture Overview](#architecture-overview) +3. [Building a Chat Application](#building-a-chat-application) +4. [Building a File Transfer System](#building-a-file-transfer-system) +5. [Building a Streaming Data Pipeline](#building-a-streaming-data-pipeline) +6. [Building a Micropython IoT Device](#building-a-micropython-iot-device) +7. [Building a Cross-Platform Dashboard](#building-a-cross-platform-dashboard) +8. [Performance Optimization](#performance-optimimization) +9. [Best Practices](#best-practices) + +--- + +## Introduction + +This walkthrough will guide you through building several real-world applications using NATSBridge. We'll cover: + +- Chat applications with rich media support +- File transfer systems with claim-check pattern +- Streaming data pipelines +- Micropython IoT devices +- Cross-platform dashboards + +Each section builds on the previous one, gradually increasing in complexity. + +--- + +## Architecture Overview + +### System Components + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ NATSBridge Architecture │ +├─────────────────────────────────────────────────────────────────┤ +│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ +│ │ Julia │ │ JavaScript │ │ Python/Micr │ │ +│ │ (NATS.jl) │◄──►│ (nats.js) │◄──►│ opython │ │ +│ └──────────────┘ └──────────────┘ └──────────────┘ │ +│ │ │ │ │ +│ ▼ ▼ ▼ │ +│ ┌──────────────────────────────────────────────┐ │ +│ │ NATS │ │ +│ │ (Message Broker) │ │ +│ └──────────────────────────────────────────────┘ │ +│ │ │ +│ ▼ │ +│ ┌──────────────────┐ │ +│ │ File Server │ │ +│ │ (HTTP Upload) │ │ +│ └──────────────────┘ │ +└─────────────────────────────────────────────────────────────────┘ +``` + +### Message Flow + +1. **Sender** creates a message envelope with payloads +2. **NATSBridge** serializes and encodes payloads +3. **Transport Decision**: Small payloads go directly to NATS, large payloads are uploaded to file server +4. **NATS** routes messages to subscribers +5. **Receiver** fetches payloads (from NATS or file server) +6. **NATSBridge** deserializes and decodes payloads + +--- + +## Building a Chat Application + +Let's build a full-featured chat application that supports text, images, and file attachments. + +### Step 1: Set Up the Project + +```bash +# Create project directory +mkdir -p chat-app/src +cd chat-app + +# Create configuration file +cat > config.json << 'EOF' +{ + "nats_url": "nats://localhost:4222", + "fileserver_url": "http://localhost:8080", + "size_threshold": 1048576 +} +EOF +``` + +### Step 2: Create the Chat Interface (JavaScript) + +```javascript +// src/chat-ui.js +class ChatUI { + constructor() { + this.messages = []; + this.currentRoom = null; + this.setupEventListeners(); + } + + setupEventListeners() { + document.getElementById('send-btn').addEventListener('click', () => this.sendMessage()); + document.getElementById('file-input').addEventListener('change', (e) => this.handleFileSelect(e)); + } + + async sendMessage() { + const messageInput = document.getElementById('message-input'); + const text = messageInput.value.trim(); + + if (!text && !this.selectedFile) return; + + const data = []; + + // Add text message + if (text) { + data.push({ + dataname: "text", + data: text, + type: "text" + }); + } + + // Add file if selected + if (this.selectedFile) { + const fileData = await this.readFile(this.selectedFile); + data.push({ + dataname: "attachment", + data: fileData, + type: this.getFileType(this.selectedFile.type) + }); + } + + await smartsend( + `/chat/${this.currentRoom}`, + data, + { + fileserverUrl: window.config.fileserver_url, + sizeThreshold: window.config.size_threshold + } + ); + + messageInput.value = ''; + this.selectedFile = null; + document.getElementById('file-name').textContent = ''; + } + + handleFileSelect(event) { + this.selectedFile = event.target.files[0]; + document.getElementById('file-name').textContent = `Selected: ${this.selectedFile.name}`; + } + + readFile(file) { + return new Promise((resolve, reject) => { + const reader = new FileReader(); + reader.onload = () => resolve(reader.result); + reader.onerror = reject; + reader.readAsArrayBuffer(file); + }); + } + + getFileType(mimeType) { + if (mimeType.startsWith('image/')) return 'image'; + if (mimeType.startsWith('audio/')) return 'audio'; + if (mimeType.startsWith('video/')) return 'video'; + return 'binary'; + } + + addMessage(user, text, attachment = null) { + const messagesContainer = document.getElementById('messages'); + const messageDiv = document.createElement('div'); + messageDiv.className = 'message'; + + let content = `
${user}
`; + content += `
${text}
`; + + if (attachment) { + if (attachment.type === 'image') { + content += ``; + } else { + content += `Download attachment`; + } + } + + messageDiv.innerHTML = content; + messagesContainer.appendChild(messageDiv); + messagesContainer.scrollTop = messagesContainer.scrollHeight; + } +} +``` + +### Step 3: Create the Message Handler + +```javascript +// src/chat-handler.js +const { smartreceive } = require('./NATSBridge'); + +class ChatHandler { + constructor(natsConnection) { + this.nats = natsConnection; + this.ui = new ChatUI(); + } + + async start() { + // Subscribe to chat rooms + const rooms = ['general', 'tech', 'random']; + + for (const room of rooms) { + await this.nats.subscribe(`/chat/${room}`, async (msg) => { + await this.handleMessage(msg); + }); + } + + console.log('Chat handler started'); + } + + async handleMessage(msg) { + const envelope = await smartreceive(msg, { + fileserverDownloadHandler: this.downloadFile.bind(this) + }); + + // Extract sender info from envelope + const sender = envelope.senderName || 'Anonymous'; + + // Process each payload + for (const payload of envelope.payloads) { + if (payload.type === 'text') { + this.ui.addMessage(sender, payload.data); + } else if (payload.type === 'image') { + // Convert to data URL for display + const base64 = this.arrayBufferToBase64(payload.data); + this.ui.addMessage(sender, null, { + type: 'image', + data: `data:image/png;base64,${base64}` + }); + } else { + // For other types, use file server URL + this.ui.addMessage(sender, null, { + type: payload.type, + data: payload.data + }); + } + } + } + + downloadFile(url, max_retries, base_delay, max_delay, correlation_id) { + return fetch(url) + .then(response => response.arrayBuffer()); + } + + arrayBufferToBase64(buffer) { + const bytes = new Uint8Array(buffer); + let binary = ''; + for (let i = 0; i < bytes.length; i++) { + binary += String.fromCharCode(bytes[i]); + } + return btoa(binary); + } +} +``` + +### Step 4: Run the Application + +```bash +# Start NATS +docker run -p 4222:4222 nats:latest + +# Start file server +mkdir -p /tmp/fileserver +python3 -m http.server 8080 --directory /tmp/fileserver + +# Run chat app +node src/chat-ui.js +node src/chat-handler.js +``` + +--- + +## Building a File Transfer System + +Let's build a file transfer system that handles large files efficiently. + +### Step 1: File Upload Service + +```javascript +// src/file-upload-service.js +const { smartsend } = require('./NATSBridge'); + +class FileUploadService { + constructor(natsUrl, fileserverUrl) { + this.natsUrl = natsUrl; + this.fileserverUrl = fileserverUrl; + } + + async uploadFile(filePath, recipient) { + const fs = require('fs'); + const fileData = fs.readFileSync(filePath); + const fileName = require('path').basename(filePath); + + const data = [{ + dataname: fileName, + data: fileData, + type: 'binary' + }]; + + const envelope = await smartsend( + `/files/${recipient}`, + data, + { + natsUrl: this.natsUrl, + fileserverUrl: this.fileserverUrl, + sizeThreshold: 1048576 + } + ); + + return envelope; + } + + async uploadLargeFile(filePath, recipient) { + // For very large files, stream upload + const fs = require('fs'); + const fileSize = fs.statSync(filePath).size; + + if (fileSize > 100 * 1024 * 1024) { // > 100MB + console.log('File too large for direct upload, using streaming...'); + return await this.streamUpload(filePath, recipient); + } + + return await this.uploadFile(filePath, recipient); + } + + async streamUpload(filePath, recipient) { + // Implement streaming upload to file server + // This would require a more sophisticated file server + // For now, we'll use the standard upload + return await this.uploadFile(filePath, recipient); + } +} + +module.exports = FileUploadService; +``` + +### Step 2: File Download Service + +```javascript +// src/file-download-service.js +const { smartreceive } = require('./NATSBridge'); +const fs = require('fs'); + +class FileDownloadService { + constructor(natsUrl) { + this.natsUrl = natsUrl; + this.downloads = new Map(); + } + + async downloadFile(sender, downloadId) { + // Subscribe to sender's file channel + const envelope = await smartreceive(msg, { + fileserverDownloadHandler: this.fetchFromUrl.bind(this) + }); + + // Process each payload + for (const payload of envelope.payloads) { + if (payload.type === 'binary') { + const filePath = `/downloads/${payload.dataname}`; + fs.writeFileSync(filePath, payload.data); + console.log(`File saved to ${filePath}`); + } + } + } + + async fetchFromUrl(url, max_retries, base_delay, max_delay, correlation_id) { + const response = await fetch(url); + if (!response.ok) { + throw new Error(`Failed to fetch: ${response.status}`); + } + return await response.arrayBuffer(); + } +} + +module.exports = FileDownloadService; +``` + +### Step 3: File Transfer CLI + +```javascript +// src/cli.js +const { smartsend, smartreceive } = require('./NATSBridge'); +const fs = require('fs'); +const readline = require('readline'); + +const rl = readline.createInterface({ + input: process.stdin, + output: process.stdout +}); + +async function main() { + const config = JSON.parse(fs.readFileSync('config.json', 'utf8')); + + console.log('File Transfer System'); + console.log('===================='); + console.log('1. Upload file'); + console.log('2. Download file'); + console.log('3. List pending downloads'); + + const choice = await rl.question('Enter choice: '); + + if (choice === '1') { + await uploadFile(config); + } else if (choice === '2') { + await downloadFile(config); + } + + rl.close(); +} + +async function uploadFile(config) { + const filePath = await rl.question('Enter file path: '); + const recipient = await rl.question('Enter recipient: '); + + const fileService = new FileUploadService(config.nats_url, config.fileserver_url); + + try { + const envelope = await fileService.uploadFile(filePath, recipient); + console.log('Upload successful!'); + console.log(`File ID: ${envelope.payloads[0].id}`); + } catch (error) { + console.error('Upload failed:', error.message); + } +} + +async function downloadFile(config) { + const sender = await rl.question('Enter sender: '); + const fileService = new FileDownloadService(config.nats_url); + + try { + await fileService.downloadFile(sender); + console.log('Download complete!'); + } catch (error) { + console.error('Download failed:', error.message); + } +} + +main(); +``` + +--- + +## Building a Streaming Data Pipeline + +Let's build a data pipeline that processes streaming data from sensors. + +### Step 1: Sensor Data Model + +```python +# src/sensor_data.py +from dataclasses import dataclass +from datetime import datetime +from typing import List, Dict, Any +import json + +@dataclass +class SensorReading: + sensor_id: str + timestamp: str + value: float + unit: str + metadata: Dict[str, Any] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "sensor_id": self.sensor_id, + "timestamp": self.timestamp, + "value": self.value, + "unit": self.unit, + "metadata": self.metadata or {} + } + +class SensorBatch: + def __init__(self): + self.readings: List[SensorReading] = [] + + def add_reading(self, reading: SensorReading): + self.readings.append(reading) + + def to_dataframe(self): + import pandas as pd + data = [r.to_dict() for r in self.readings] + return pd.DataFrame(data) +``` + +### Step 2: Sensor Sender + +```python +# src/sensor-sender.py +from nats_bridge import smartsend +from sensor_data import SensorReading, SensorBatch +import time +import random + +class SensorSender: + def __init__(self, nats_url: str, fileserver_url: str): + self.nats_url = nats_url + self.fileserver_url = fileserver_url + + def send_reading(self, sensor_id: str, value: float, unit: str): + reading = SensorReading( + sensor_id=sensor_id, + timestamp=datetime.now().isoformat(), + value=value, + unit=unit + ) + + data = [("reading", reading.to_dict(), "dictionary")] + + smartsend( + f"/sensors/{sensor_id}", + data, + nats_url=self.nats_url, + fileserver_url=self.fileserver_url + ) + + def send_batch(self, readings: List[SensorReading]): + batch = SensorBatch() + for reading in readings: + batch.add_reading(reading) + + df = batch.to_dataframe() + + # Convert to Arrow IPC format + import pyarrow as pa + table = pa.Table.from_pandas(df) + + # Serialize to Arrow IPC + import io + buf = io.BytesIO() + with pa.ipc.new_stream(buf, table.schema) as writer: + writer.write_table(table) + + arrow_data = buf.getvalue() + + # Send based on size + if len(arrow_data) < 1048576: # < 1MB + data = [("batch", arrow_data, "table")] + smartsend( + f"/sensors/batch", + data, + nats_url=self.nats_url, + fileserver_url=self.fileserver_url + ) + else: + # Upload to file server + from nats_bridge import plik_oneshot_upload + response = plik_oneshot_upload(self.fileserver_url, "batch.arrow", arrow_data) + print(f"Uploaded batch to {response['url']}") +``` + +### Step 3: Sensor Receiver + +```python +# src/sensor-receiver.py +from nats_bridge import smartreceive +from sensor_data import SensorReading +import pandas as pd +import pyarrow as pa +import io + +class SensorReceiver: + def __init__(self, fileserver_download_handler): + self.fileserver_download_handler = fileserver_download_handler + + def process_reading(self, msg): + envelope = smartreceive(msg, self.fileserver_download_handler) + + for dataname, data, data_type in envelope["payloads"]: + if data_type == "dictionary": + reading = SensorReading( + sensor_id=data["sensor_id"], + timestamp=data["timestamp"], + value=data["value"], + unit=data["unit"], + metadata=data.get("metadata", {}) + ) + print(f"Received: {reading}") + + elif data_type == "table": + # Deserialize Arrow IPC + table = pa.ipc.open_stream(io.BytesIO(data)).read_all() + df = table.to_pandas() + print(f"Received batch with {len(df)} readings") + print(df) +``` + +--- + +## Building a Micropython IoT Device + +Let's build an IoT device using Micropython that connects to NATS. + +### Step 1: Device Configuration + +```python +# device_config.py +import json + +class DeviceConfig: + def __init__(self, ssid, password, nats_url, device_id): + self.ssid = ssid + self.password = password + self.nats_url = nats_url + self.device_id = device_id + + def to_dict(self): + return { + "ssid": self.ssid, + "password": self.password, + "nats_url": self.nats_url, + "device_id": self.device_id + } +``` + +### Step 2: Device Bridge + +```python +# device_bridge.py +from nats_bridge import smartsend, smartreceive +import json + +class DeviceBridge: + def __init__(self, config): + self.config = config + self.nats_url = config.nats_url + + def connect(self): + # Connect to WiFi + import network + wlan = network.WLAN(network.STA_IF) + wlan.active(True) + wlan.connect(self.config.ssid, self.config.password) + + while not wlan.isconnected(): + pass + + print('Connected to WiFi') + print('Network config:', wlan.ifconfig()) + + def send_status(self, status): + data = [("status", status, "dictionary")] + smartsend( + f"/devices/{self.config.device_id}/status", + data, + nats_url=self.nats_url + ) + + def send_sensor_data(self, sensor_id, value, unit): + data = [ + ("sensor_id", sensor_id, "text"), + ("value", value, "dictionary") + ] + smartsend( + f"/devices/{self.config.device_id}/sensors/{sensor_id}", + data, + nats_url=self.nats_url + ) + + def receive_commands(self, callback): + # Subscribe to commands + import socket + # Simplified subscription - in production, use proper NATS client + + while True: + # Poll for messages + msg = self._poll_for_message() + if msg: + envelope = smartreceive(msg) + + # Process payloads + for dataname, data, data_type in envelope["payloads"]: + if dataname == "command": + callback(data) + + def _poll_for_message(self): + # Simplified message polling + # In production, implement proper NATS subscription + return None +``` + +### Step 3: Device Application + +```python +# device_app.py +from device_config import DeviceConfig +from device_bridge import DeviceBridge +import time +import random + +# Load configuration +config = DeviceConfig( + ssid="MyNetwork", + password="password123", + nats_url="nats://localhost:4222", + device_id="device-001" +) + +bridge = DeviceBridge(config) +bridge.connect() + +# Send initial status +bridge.send_status({ + "status": "online", + "version": "1.0.0", + "uptime": 0 +}) + +# Main loop +while True: + # Read sensor data + temperature = random.uniform(20, 30) + humidity = random.uniform(40, 60) + + bridge.send_sensor_data("temperature", temperature, "celsius") + bridge.send_sensor_data("humidity", humidity, "percent") + + # Send status update + bridge.send_status({ + "status": "online", + "temperature": temperature, + "humidity": humidity + }) + + time.sleep(60) # Every minute +``` + +--- + +## Building a Cross-Platform Dashboard + +Let's build a dashboard that displays data from multiple platforms. + +### Step 1: Dashboard Server (Python) + +```python +# src/dashboard-server.py +from nats_bridge import smartsend, smartreceive +import pandas as pd +import pyarrow as pa +import io + +class DashboardServer: + def __init__(self, nats_url, fileserver_url): + self.nats_url = nats_url + self.fileserver_url = fileserver_url + + def broadcast_data(self, df): + # Convert to Arrow IPC + table = pa.Table.from_pandas(df) + buf = io.BytesIO() + with pa.ipc.new_stream(buf, table.schema) as writer: + writer.write_table(table) + + arrow_data = buf.getvalue() + + # Broadcast to all subscribers + data = [("data", arrow_data, "table")] + smartsend( + "/dashboard/data", + data, + nats_url=self.nats_url, + fileserver_url=self.fileserver_url + ) + + def receive_selection(self, callback): + def handler(msg): + envelope = smartreceive(msg) + + for dataname, data, data_type in envelope["payloads"]: + if data_type == "dictionary": + callback(data) + + # Subscribe to selections + import threading + thread = threading.Thread(target=self._listen_for_selections, args=(handler,)) + thread.daemon = True + thread.start() + + def _listen_for_selections(self, handler): + # Simplified subscription + # In production, implement proper NATS subscription + pass +``` + +### Step 2: Dashboard UI (JavaScript) + +```javascript +// src/dashboard-ui.js +class DashboardUI { + constructor() { + this.data = null; + this.setupEventListeners(); + } + + setupEventListeners() { + document.getElementById('refresh-btn').addEventListener('click', () => this.refreshData()); + document.getElementById('export-btn').addEventListener('click', () => this.exportData()); + } + + async refreshData() { + // Request fresh data + await smartsend("/dashboard/request", [ + { dataname: "request", data: { type: "refresh" }, type: "dictionary" } + ], { + fileserverUrl: window.config.fileserver_url + }); + } + + async fetchData() { + // Subscribe to data updates + const envelope = await smartreceive(msg, { + fileserverDownloadHandler: this.fetchFromUrl.bind(this) + }); + + // Process table data + for (const payload of envelope.payloads) { + if (payload.type === 'table') { + // Deserialize Arrow IPC + this.data = this.deserializeArrow(payload.data); + this.renderTable(); + } + } + } + + deserializeArrow(data) { + // Deserialize Arrow IPC to JavaScript array + // This would require the apache-arrow library + return JSON.parse(JSON.stringify(data)); // Simplified + } + + renderTable() { + const tableContainer = document.getElementById('data-table'); + tableContainer.innerHTML = ''; + + if (!this.data) return; + + // Render table headers + const headers = Object.keys(this.data[0]); + const thead = document.createElement('thead'); + const headerRow = document.createElement('tr'); + + headers.forEach(header => { + const th = document.createElement('th'); + th.textContent = header; + headerRow.appendChild(th); + }); + + thead.appendChild(headerRow); + tableContainer.appendChild(thead); + + // Render table rows + const tbody = document.createElement('tbody'); + + this.data.forEach(row => { + const tr = document.createElement('tr'); + headers.forEach(header => { + const td = document.createElement('td'); + td.textContent = row[header]; + tr.appendChild(td); + }); + tbody.appendChild(tr); + }); + + tableContainer.appendChild(tbody); + } + + exportData() { + const csv = this.toCSV(); + const blob = new Blob([csv], { type: 'text/csv' }); + const url = URL.createObjectURL(blob); + + const a = document.createElement('a'); + a.href = url; + a.download = 'dashboard_data.csv'; + a.click(); + } + + toCSV() { + if (!this.data) return ''; + + const headers = Object.keys(this.data[0]); + const rows = this.data.map(row => + headers.map(h => row[h]).join(',') + ); + + return [headers.join(','), ...rows].join('\n'); + } + + fetchFromUrl(url) { + return fetch(url) + .then(response => response.arrayBuffer()); + } +} +``` + +--- + +## Performance Optimization + +### 1. Batch Processing + +```python +# Batch multiple readings into a single message +def send_batch_readings(self, readings): + batch = SensorBatch() + for reading in readings: + batch.add_reading(reading) + + df = batch.to_dataframe() + + # Convert to Arrow IPC + table = pa.Table.from_pandas(df) + buf = io.BytesIO() + with pa.ipc.new_stream(buf, table.schema) as writer: + writer.write_table(table) + + arrow_data = buf.getvalue() + + # Send as single message + smartsend( + "/sensors/batch", + [("batch", arrow_data, "table")], + nats_url=self.nats_url + ) +``` + +### 2. Connection Pooling + +```javascript +// Reuse NATS connections +const nats = require('nats'); + +class ConnectionPool { + constructor() { + this.connections = new Map(); + } + + getConnection(natsUrl) { + if (!this.connections.has(natsUrl)) { + this.connections.set(natsUrl, nats.connect({ servers: [natsUrl] })); + } + return this.connections.get(natsUrl); + } + + closeAll() { + this.connections.forEach(conn => conn.close()); + this.connections.clear(); + } +} +``` + +### 3. Caching + +```python +# Cache file server responses +from functools import lru_cache + +@lru_cache(maxsize=100) +def fetch_with_caching(url, max_retries=5, base_delay=100, max_delay=5000, correlation_id=""): + return _fetch_with_backoff(url, max_retries, base_delay, max_delay, correlation_id) +``` + +--- + +## Best Practices + +### 1. Error Handling + +```python +def safe_smartsend(subject, data, **kwargs): + try: + return smartsend(subject, data, **kwargs) + except Exception as e: + print(f"Failed to send message: {e}") + return None +``` + +### 2. Logging + +```python +import logging + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +def log_send(subject, data, correlation_id): + logger.info(f"Sending to {subject}: {len(data)} payloads, correlation_id={correlation_id}") + +def log_receive(correlation_id, num_payloads): + logger.info(f"Received message: {num_payloads} payloads, correlation_id={correlation_id}") +``` + +### 3. Rate Limiting + +```python +import time +from collections import deque + +class RateLimiter: + def __init__(self, max_requests, time_window): + self.max_requests = max_requests + self.time_window = time_window + self.requests = deque() + + def allow(self): + now = time.time() + + # Remove old requests + while self.requests and self.requests[0] < now - self.time_window: + self.requests.popleft() + + if len(self.requests) >= self.max_requests: + return False + + self.requests.append(now) + return True +``` + +--- + +## Conclusion + +This walkthrough covered: + +- Building a chat application with rich media support +- Building a file transfer system with claim-check pattern +- Building a streaming data pipeline for sensor data +- Building a Micropython IoT device +- Building a cross-platform dashboard + +For more information, check the [API documentation](../src/README.md) and [test examples](../test/). + +--- + +## License + +MIT \ No newline at end of file diff --git a/examples/walkthrough_julia.md b/examples/walkthrough_julia.md deleted file mode 100644 index d7081dc..0000000 --- a/examples/walkthrough_julia.md +++ /dev/null @@ -1,939 +0,0 @@ -# NATSBridge.jl Walkthrough: Building a Chat System - -A step-by-step guided walkthrough for building a real-time chat system using NATSBridge.jl with mixed content support (text, images, audio, video, and files). - -## Prerequisites - -- Julia 1.7+ -- NATS server running -- HTTP file server (Plik) running - -## Step 1: Understanding the Chat System Architecture - -### System Components - -``` -┌─────────────────────────────────────────────────────────────────────────────┐ -│ Chat System │ -├─────────────────────────────────────────────────────────────────────────────┤ -│ │ -│ ┌──────────────┐ NATS ┌──────────────┐ │ -│ │ Julia │◄───────┬───────► │ JavaScript │ │ -│ │ Service │ │ │ Client │ │ -│ │ │ │ │ │ │ -│ │ - Text │ │ │ - Text │ │ -│ │ - Images │ │ │ - Images │ │ -│ │ - Audio │ ▼ │ - Audio │ │ -│ │ - Video │ NATSBridge.jl │ - Files │ │ -│ │ - Files │ │ │ - Tables │ │ -│ └──────────────┘ │ └──────────────┘ │ -│ │ │ -│ ┌───────┴───────┐ │ -│ │ NATS │ │ -│ │ Server │ │ -│ └─────────────┘ │ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ - -For large payloads (> 1MB): -┌─────────────────────────────────────────────────────────────────────────────┐ -│ File Server (Plik) │ -│ │ -│ Julia Service ──► Upload ──► File Server ──► Download ◄── JavaScript Client│ -│ │ -└─────────────────────────────────────────────────────────────────────────────┘ -``` - -### Message Format - -Each chat message is an envelope containing multiple payloads: - -```json -{ - "correlationId": "uuid4", - "msgId": "uuid4", - "timestamp": "2024-01-15T10:30:00Z", - "sendTo": "/chat/room1", - "msgPurpose": "chat", - "senderName": "user-1", - "senderId": "uuid4", - "receiverName": "user-2", - "receiverId": "uuid4", - "brokerURL": "nats://localhost:4222", - "payloads": [ - { - "id": "uuid4", - "dataname": "message_text", - "type": "text", - "transport": "direct", - "encoding": "base64", - "size": 256, - "data": "SGVsbG8gV29ybGQh", - "metadata": {} - }, - { - "id": "uuid4", - "dataname": "user_image", - "type": "image", - "transport": "link", - "encoding": "none", - "size": 15433, - "data": "http://localhost:8080/file/UPLOAD_ID/FILE_ID/image.jpg", - "metadata": {} - } - ] -} -``` - -## Step 2: Setting Up the Environment - -### 1. Start NATS Server - -```bash -# Using Docker -docker run -d -p 4222:4222 -p 8222:8222 --name nats-server nats:latest - -# Or download from https://github.com/nats-io/nats-server/releases -./nats-server -``` - -### 2. Start HTTP File Server (Plik) - -```bash -# Using Docker -docker run -d -p 8080:8080 --name plik plik/plik:latest - -# Or download from https://github.com/arnaud-lb/plik/releases -./plikd -d -``` - -### 3. Install Julia Dependencies - -```julia -using Pkg -Pkg.add("NATS") -Pkg.add("JSON") -Pkg.add("Arrow") -Pkg.add("HTTP") -Pkg.add("UUIDs") -Pkg.add("Dates") -Pkg.add("Base64") -Pkg.add("PrettyPrinting") -Pkg.add("DataFrames") -``` - -## Step 3: Basic Text-Only Chat - -### Sender (User 1) - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -# Include the bridge module -include("NATSBridge.jl") -using .NATSBridge - -# Configuration -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -# File upload handler for plik server -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -# Send a simple text message -function send_text_message() - message_text = "Hello, how are you today?" - - env = NATSBridge.smartsend( - SUBJECT, - [("message", message_text, "text")], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Sent text message with correlation ID: $(env.correlationId)") -end - -send_text_message() -``` - -### Receiver (User 2) - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -# Include the bridge module -include("NATSBridge.jl") -using .NATSBridge - -# Configuration -const NATS_URL = "nats://localhost:4222" -const SUBJECT = "/chat/room1" - -# Message handler -function message_handler(msg::NATS.Msg) - payloads = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - # Extract the text message - for (dataname, data, data_type) in payloads - if data_type == "text" - println("Received message: $data") - # Save to file - write("./received_$dataname.txt", data) - end - end -end - -# Subscribe to the chat room -NATS.subscribe(SUBJECT) do msg - message_handler(msg) -end - -# Keep the program running -while true - sleep(1) -end -``` - -## Step 4: Adding Image Support - -### Sending an Image - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_image() - # Read image file - image_data = read("screenshot.png", Vector{UInt8}) - - # Send with text message - env = NATSBridge.smartsend( - SUBJECT, - [ - ("text", "Check out this screenshot!", "text"), - ("screenshot", image_data, "image") - ], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Sent image with correlation ID: $(env.correlationId)") -end - -send_image() -``` - -### Receiving an Image - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const SUBJECT = "/chat/room1" - -function message_handler(msg::NATS.Msg) - payloads = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in payloads - if data_type == "text" - println("Text: $data") - elseif data_type == "image" - # Save image to file - filename = "received_$dataname.bin" - write(filename, data) - println("Saved image: $filename") - end - end -end - -NATS.subscribe(SUBJECT) do msg - message_handler(msg) -end -``` - -## Step 5: Handling Large Files with Link Transport - -### Automatic Transport Selection - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_large_file() - # Create a large file (> 1MB triggers link transport) - large_data = rand(10_000_000) # ~80MB - - env = NATSBridge.smartsend( - SUBJECT, - [("large_file", large_data, "binary")], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Uploaded large file to: $(env.payloads[1].data)") - println("Correlation ID: $(env.correlationId)") -end - -send_large_file() -``` - -## Step 6: Audio and Video Support - -### Sending Audio - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_audio() - # Read audio file (WAV, MP3, etc.) - audio_data = read("voice_message.mp3", Vector{UInt8}) - - env = NATSBridge.smartsend( - SUBJECT, - [("voice_message", audio_data, "audio")], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Sent audio message: $(env.correlationId)") -end - -send_audio() -``` - -### Sending Video - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_video() - # Read video file (MP4, AVI, etc.) - video_data = read("video_message.mp4", Vector{UInt8}) - - env = NATSBridge.smartsend( - SUBJECT, - [("video_message", video_data, "video")], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Sent video message: $(env.correlationId)") -end - -send_video() -``` - -## Step 7: Table/Data Exchange - -### Sending Tabular Data - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_table() - # Create a DataFrame - df = DataFrame( - id = 1:5, - name = ["Alice", "Bob", "Charlie", "Diana", "Eve"], - score = [95, 88, 92, 98, 85], - grade = ['A', 'B', 'A', 'B', 'B'] - ) - - env = NATSBridge.smartsend( - SUBJECT, - [("student_scores", df, "table")], - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Sent table with $(nrow(df)) rows") -end - -send_table() -``` - -### Receiving and Using Tables - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const SUBJECT = "/chat/room1" - -function message_handler(msg::NATS.Msg) - payloads = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in payloads - if data_type == "table" - data = DataFrame(data) - println("Received table:") - show(data) - println("\nAverage score: $(mean(data.score))") - end - end -end - -NATS.subscribe(SUBJECT) do msg - message_handler(msg) -end -``` - -## Step 8: Bidirectional Communication - -### Request-Response Pattern - -```julia -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -include("NATSBridge.jl") -using .NATSBridge - -const NATS_URL = "nats://localhost:4222" -const SUBJECT = "/api/query" -const REPLY_SUBJECT = "/api/response" - -# Request -function send_request() - query_data = Dict("query" => "SELECT * FROM users") - - env = NATSBridge.smartsend( - SUBJECT, - [("sql_query", query_data, "dictionary")], - nats_url = NATS_URL, - fileserver_url = "http://localhost:8080", - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "request", - sender_name = "frontend", - receiver_name = "backend", - reply_to = REPLY_SUBJECT, - reply_to_msg_id = string(uuid4()) - ) - - println("Request sent: $(env.correlationId)") -end - -# Response handler -function response_handler(msg::NATS.Msg) - payloads = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - for (dataname, data, data_type) in payloads - if data_type == "table" - data = DataFrame(data) - println("Query results:") - show(data) - end - end -end - -NATS.subscribe(REPLY_SUBJECT) do msg - response_handler(msg) -end -``` - -## Step 9: Complete Chat Application - -### Full Chat System - -```julia -module ChatApp -using NATS -using JSON -using UUIDs -using Dates -using PrettyPrinting -using DataFrames -using Arrow -using HTTP -using Base64 - -# Include the bridge module -include("../src/NATSBridge.jl") -using .NATSBridge - -# Configuration -const NATS_URL = "nats://localhost:4222" -const FILESERVER_URL = "http://localhost:8080" -const SUBJECT = "/chat/room1" - -# File upload handler for plik server -function plik_upload_handler(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any} - url_getUploadID = "$fileserver_url/upload" - headers = ["Content-Type" => "application/json"] - body = """{ "OneShot" : true }""" - httpResponse = HTTP.request("POST", url_getUploadID, headers, body; body_is_form=false) - responseJson = JSON.parse(String(httpResponse.body)) - uploadid = responseJson["id"] - uploadtoken = responseJson["uploadToken"] - - file_multipart = HTTP.Multipart(dataname, IOBuffer(data), "application/octet-stream") - url_upload = "$fileserver_url/file/$uploadid" - headers = ["X-UploadToken" => uploadtoken] - - form = HTTP.Form(Dict("file" => file_multipart)) - httpResponse = HTTP.post(url_upload, headers, form) - responseJson = JSON.parse(String(httpResponse.body)) - - fileid = responseJson["id"] - url = "$fileserver_url/file/$uploadid/$fileid/$dataname" - - return Dict("status" => httpResponse.status, "uploadid" => uploadid, "fileid" => fileid, "url" => url) -end - -function send_chat_message( - text::String, - image_path::Union{String, Nothing}=nothing, - audio_path::Union{String, Nothing}=nothing -) - # Build payloads list - payloads = [("message_text", text, "text")] - - if image_path !== nothing - image_data = read(image_path, Vector{UInt8}) - push!(payloads, ("user_image", image_data, "image")) - end - - if audio_path !== nothing - audio_data = read(audio_path, Vector{UInt8}) - push!(payloads, ("user_audio", audio_data, "audio")) - end - - env = NATSBridge.smartsend( - SUBJECT, - payloads, - nats_url = NATS_URL, - fileserver_url = FILESERVER_URL, - fileserverUploadHandler = plik_upload_handler, - size_threshold = 1_000_000, - correlation_id = string(uuid4()), - msg_purpose = "chat", - sender_name = "user-1" - ) - - println("Message sent with correlation ID: $(env.correlationId)") -end - -function receive_chat_messages() - function message_handler(msg::NATS.Msg) - payloads = NATSBridge.smartreceive( - msg, - max_retries = 5, - base_delay = 100, - max_delay = 5000 - ) - - println("\n--- New Message ---") - for (dataname, data, data_type) in payloads - if data_type == "text" - println("Text: $data") - elseif data_type == "image" - filename = "received_$dataname.bin" - write(filename, data) - println("Image saved: $filename") - elseif data_type == "audio" - filename = "received_$dataname.bin" - write(filename, data) - println("Audio saved: $filename") - elseif data_type == "table" - println("Table received:") - data = DataFrame(data) - show(data) - end - end - end - - NATS.subscribe(SUBJECT) do msg - message_handler(msg) - end - println("Subscribed to: $SUBJECT") -end - -function run_interactive_chat() - println("\n=== Interactive Chat ===") - println("1. Send a message") - println("2. Join a chat room") - println("3. Exit") - - while true - print("\nSelect option (1-3): ") - choice = readline() - - if choice == "1" - print("Enter message text: ") - text = readline() - send_chat_message(text) - elseif choice == "2" - receive_chat_messages() - elseif choice == "3" - break - end - end -end - -end # module - -# Run the chat app -using .ChatApp -ChatApp.run_interactive_chat() -``` - -## Step 10: Testing the Chat System - -### Test Scenario 1: Text-Only Chat - -```bash -# Terminal 1: Start the chat receiver -julia test_julia_to_julia_text_receiver.jl - -# Terminal 2: Send a message -julia test_julia_to_julia_text_sender.jl -``` - -### Test Scenario 2: Image Chat - -```bash -# Terminal 1: Receive messages -julia test_julia_to_julia_mix_payloads_receiver.jl - -# Terminal 2: Send image -julia test_julia_to_julia_mix_payload_sender.jl -``` - -### Test Scenario 3: Large File Transfer - -```bash -# Terminal 2: Send large file -julia test_julia_to_julia_mix_payload_sender.jl -``` - -## Conclusion - -This walkthrough demonstrated how to build a chat system using NATSBridge.jl with support for: - -- Text messages -- Images (direct transport for small, link transport for large) -- Audio files -- Video files -- Tabular data (DataFrames) -- Bidirectional communication -- Mixed-content messages - -The key takeaways are: - -1. **Always wrap payloads in a list** - Even for single payloads: `[("dataname", data, "type")]` -2. **Use appropriate transport** - NATSBridge automatically handles size-based routing -3. **Support mixed content** - Multiple payloads of different types in one message -4. **Handle errors** - Implement proper error handling for network failures -5. **Use correlation IDs** - Track messages across distributed systems - -For more information, see: -- [`docs/architecture.md`](./docs/architecture.md) - Detailed architecture documentation -- [`docs/implementation.md`](./docs/implementation.md) - Implementation details \ No newline at end of file