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

update

Browse Source
This commit is contained in:
narawat
2026-05-25 10:32:12 +07:00
parent 48fddb5cbc
commit da2085ad32
1 changed files with 160 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

204
README.md
View File

@@ -50,6 +50,121 @@ A high-performance, **transport-agnostic** communication layer for both **Julia*
| `"video"` | `Vector{UInt8}` | Video data | | `"video"` | `Vector{UInt8}` | Video data |
| `"binary"` | `Vector{UInt8}`, `IOBuffer` | Binary data | | `"binary"` | `Vector{UInt8}`, `IOBuffer` | Binary data |
### Message Envelope Structure (JSON) with Multiple Payloads
After calling `smartpack()`, the data is serialized into a JSON message envelope with the following structure:
```json
{
"correlation_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"msg_id": "b2c3d4e5-f6a7-8901-bcde-f12345678901",
"timestamp": "2026-05-25T08:12:40.000Z",
"send_to": "/chat/room1",
"msg_purpose": "chat",
"sender_name": "msghandler",
"sender_id": "c3d4e5f6-a7b8-9012-cdef-123456789012",
"receiver_name": "",
"receiver_id": "",
"reply_to": "",
"reply_to_msg_id": "",
"broker_url": "nats.yiem.cc",
"metadata": {},
"payloads": [
{
"id": "d4e5f6a7-b8c9-0123-defa-234567890123",
"dataname": "message",
"payload_type": "text",
"transport": "direct",
"encoding": "base64",
"size": 6,
"data": "SGVsbG8h",
"metadata": {}
},
{
"id": "e5f6a7b8-c9d0-1234-efab-345678901234",
"dataname": "config",
"payload_type": "dictionary",
"transport": "direct",
"encoding": "base64",
"size": 18,
"data": "eyJrZXkiOiJ2YWx1ZSJ9",
"metadata": {}
},
{
"id": "f6a7b8c9-d0e1-2345-fabc-456789012345",
"dataname": "small_file",
"payload_type": "binary",
"transport": "direct",
"encoding": "base64",
"size": 1024,
"data": "base64encodedbinarydata...",
"metadata": {}
},
{
"id": "a7b8c9d0-e1f2-3456-abcd-567890123456",
"dataname": "large_image",
"payload_type": "image",
"transport": "link",
"encoding": "none",
"size": 2048576,
"data": "http://localhost:8080/file/ABC12/XYZ99/image.png",
"metadata": {}
}
]
}
```
#### Payload Transport Types
| Transport | Data Format | Use Case |
|-----------|-------------|----------|
| `direct` | Base64-encoded string | Payloads < 500KB sent directly |
| `link` | HTTP URL | Payloads ≥ 500KB uploaded to file server |
#### Payload Encoding Types
| Encoding | Payload Types |
|----------|---------------|
| `base64` | text, image, audio, video, binary |
| `json` | dictionary, jsontable |
| `arrow-ipc` | arrowtable |
| `none` | link transport (URL only) |
#### Payload Field Definitions
| Field | Type | Description |
|-------|------|-------------|
| `id` | `string` (UUID) | Unique payload identifier |
| `dataname` | `string` | Name/filename of the payload |
| `payload_type` | `string` | Type: text, dictionary, arrowtable, jsontable, image, audio, video, binary |
| `transport` | `string` | `direct` or `link` |
| `encoding` | `string` | base64, json, arrow-ipc, or none |
| `size` | `integer` | Size in bytes |
| `data` | `string` | Base64 string (direct) or URL (link) |
| `metadata` | `object` | Optional metadata |
#### Example: Multiple Payloads with Mixed Types
```julia
using msghandler
data = [
("message", "Hello!", "text"),
("config", Dict("key" => "value"), "dictionary"),
("small_file", file_bytes, "binary"),
("large_image", image_bytes, "image")
]
message_envelope, message_envelope_json_str = msghandler.smartpack("/chat/room1", data;
broker_url="nats.yiem.cc",
fileserver_url="http://localhost:8080"
)
# message_envelope_json_str contains the JSON message_envelope shown above
# Payloads < 500KB use 'transport': 'direct' with Base64 data
# Payloads ≥ 500KB use 'transport': 'link' with URL to file server
```
### Examples ### Examples
**Sending multiple payloads:** **Sending multiple payloads:**
@@ -57,9 +172,10 @@ A high-performance, **transport-agnostic** communication layer for both **Julia*
data = [ data = [
("message", "Hello!", "text"), ("message", "Hello!", "text"),
("config", Dict("key" => "value"), "dictionary"), ("config", Dict("key" => "value"), "dictionary"),
("file", file_bytes, "binary") ("small_file", file_bytes, "binary"),
("large_image", image_bytes, "image")
] ]
env, json_str = msghandler.smartpack("subject", data, options...) message_envelope, message_envelope_json_str = msghandler.smartpack("subject", data, options...)
``` ```
**Important Notes:** **Important Notes:**
@@ -214,7 +330,7 @@ payload_2 = (filename_large_image, file_data_large_image, "binary")
payloads = [payload_1, payload_2] # List of tuples payloads = [payload_1, payload_2] # List of tuples
# Step 1: Create the message envelope (transport-agnostic) # Step 1: Create the message envelope (transport-agnostic)
envelope, envelope_json_str = msghandler.smartpack("test.topic", message_envelope, message_envelope_json_str = msghandler.smartpack("test.topic",
payloads; payloads;
broker_url="nats.yiem.cc", broker_url="nats.yiem.cc",
fileserver_url="http://192.168.88.104:8080") fileserver_url="http://192.168.88.104:8080")
@@ -222,11 +338,11 @@ envelope, envelope_json_str = msghandler.smartpack("test.topic",
# Step 2: Send via your chosen transport (NATS in this example) # Step 2: Send via your chosen transport (NATS in this example)
using NATS using NATS
conn = NATS.connect("nats.yiem.cc") conn = NATS.connect("nats.yiem.cc")
NATS.publish(conn, "test.topic", envelope_json_str; reply_to="test.replytopic") NATS.publish(conn, "test.topic", message_envelope_json_str; reply_to="test.replytopic")
NATS.drain(conn) NATS.drain(conn)
# OR using request-reply pattern # OR using request-reply pattern
reply = NATS.request(conn, "test.topic", envelope_json_str, timeout=10) reply = NATS.request(conn, "test.topic", message_envelope_json_str, timeout=10)
``` ```
#### Receive Messages (Publish - Subscribe Pattern) #### Receive Messages (Publish - Subscribe Pattern)
@@ -237,16 +353,16 @@ using msghandler, NATS
conn = NATS.connect("nats.yiem.cc") conn = NATS.connect("nats.yiem.cc")
NATS.subscribe(conn, "test.topic") do msg NATS.subscribe(conn, "test.topic") do msg
println("Received message on $(msg.subject)") println("Received message on $(msg.subject)")
envelope_json_str = String(msg.payload) message_envelope_json_str = String(msg.payload)
# Step 2: Unpack the envelope (transport-agnostic) # Step 2: Unpack the message_envelope (transport-agnostic)
envelope = msghandler.smartunpack( message_envelope = msghandler.smartunpack(
envelope_json_str; message_envelope_json_str;
max_retries = 5, max_retries = 5,
base_delay = 100, base_delay = 100,
max_delay = 5000 max_delay = 5000
) )
println(envelope.payloads[1]) println(message_envelope.payloads[1])
end end
``` ```
@@ -266,7 +382,7 @@ payload_2 = (filename_large_image, file_data_large_image, "binary")
payloads = [payload_1, payload_2] # List of tuples payloads = [payload_1, payload_2] # List of tuples
# Step 1: Create the message envelope (transport-agnostic) # Step 1: Create the message envelope (transport-agnostic)
envelope, envelope_json_str = msghandler.smartpack("test.topic", message_envelope, message_envelope_json_str = msghandler.smartpack("test.topic",
payloads; payloads;
broker_url="nats.yiem.cc", broker_url="nats.yiem.cc",
fileserver_url="http://192.168.88.104:8080") fileserver_url="http://192.168.88.104:8080")
@@ -274,7 +390,7 @@ envelope, envelope_json_str = msghandler.smartpack("test.topic",
# Step 2: Send via your chosen transport (NATS in this example) # Step 2: Send via your chosen transport (NATS in this example)
using NATS using NATS
conn = NATS.connect("nats.yiem.cc") conn = NATS.connect("nats.yiem.cc")
reply = NATS.request(conn, "test.topic", envelope_json_str, timeout=10) reply = NATS.request(conn, "test.topic", message_envelope_json_str, timeout=10)
``` ```
#### Receive Messages (Request - Reply Pattern) #### Receive Messages (Request - Reply Pattern)
@@ -284,14 +400,14 @@ using msghandler, NATS
conn = NATS.connect("nats.yiem.cc") conn = NATS.connect("nats.yiem.cc")
sub1 = NATS.reply(conn, "nats.yiem.cc"; queue_group="group1", spawn=true) do msg sub1 = NATS.reply(conn, "nats.yiem.cc"; queue_group="group1", spawn=true) do msg
envelope_json_str = String(msg.payload) message_envelope_json_str = String(msg.payload)
envelope = msghandler.smartunpack(envelope_json_str) message_envelope = msghandler.smartunpack(message_envelope_json_str)
response = [("respond_message", "msg received", "text")] response = [("respond_message", "msg received", "text")]
_, envelope_json_str = msghandler.smartpack("requester inbox", _, response_message_envelope_json_str = msghandler.smartpack("requester inbox",
response; response;
broker_url="nats.yiem.cc", broker_url="nats.yiem.cc",
fileserver_url="http://192.168.88.104:8080") fileserver_url="http://192.168.88.104:8080")
return envelope_json_str return response_message_envelope_json_str
end end
``` ```
@@ -332,19 +448,19 @@ const payload4 = ["user_avatar", file_data, "binary"];
const payloads = [payload1, payload2, payload3, payload4]; // Array of arrays const payloads = [payload1, payload2, payload3, payload4]; // Array of arrays
// Step 1: Create the message envelope (transport-agnostic) // Step 1: Create the message envelope (transport-agnostic)
const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack("test.topic", payloads, { const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack("test.topic", payloads, {
broker_url: "nats.yiem.cc", broker_url: "nats.yiem.cc",
fileserver_url: "http://192.168.88.104:8080" fileserver_url: "http://192.168.88.104:8080"
}); });
// Step 2: Send via your chosen transport (NATS, WebSocket, HTTP, etc.) // Step 2: Send via your chosen transport (NATS, WebSocket, HTTP, etc.)
// await myTransport.publish("test.topic", envelopeJsonStr); // await myTransport.publish("test.topic", message_envelope_json_str);
// OR using fetch API for HTTP transport // OR using fetch API for HTTP transport
fetch("http://localhost:3000/api/messages", { fetch("http://localhost:3000/api/messages", {
method: "POST", method: "POST",
headers: { "Content-Type": "application/json" }, headers: { "Content-Type": "application/json" },
body: envelopeJsonStr body: message_envelope_json_str
}); });
``` ```
@@ -356,7 +472,7 @@ import msghandlerCSR from './src/msghandler-csr.js';
const { smartunpack } = msghandlerCSR; const { smartunpack } = msghandlerCSR;
// Option 1: Direct JSON string // Option 1: Direct JSON string
const envelope = await msghandlerCSR.smartunpack(jsonString, { const message_envelope = await msghandlerCSR.smartunpack(jsonString, {
fileserver_download_handler: msghandlerCSR.fetchWithBackoff, fileserver_download_handler: msghandlerCSR.fetchWithBackoff,
max_retries: 5, max_retries: 5,
base_delay: 100, base_delay: 100,
@@ -364,12 +480,12 @@ const envelope = await msghandlerCSR.smartunpack(jsonString, {
}); });
// Option 2: From transport message object (e.g., NATS, WebSocket) // Option 2: From transport message object (e.g., NATS, WebSocket)
const envelope = await msghandlerCSR.smartunpack(natsMessage, { const message_envelope = await msghandlerCSR.smartunpack(natsMessage, {
fileserver_download_handler: msghandlerCSR.fetchWithBackoff fileserver_download_handler: msghandlerCSR.fetchWithBackoff
}); });
// Process payloads // Process payloads
for (const [dataname, data, type] of envelope.payloads) { for (const [dataname, data, type] of message_envelope.payloads) {
console.log(`${dataname}:`, data, `(type: ${type})`); console.log(`${dataname}:`, data, `(type: ${type})`);
} }
``` ```
@@ -384,7 +500,7 @@ Sends data via your chosen transport mechanism with intelligent transport select
```julia ```julia
using msghandler using msghandler
env, env_json_str = msghandler.smartpack( message_envelope, message_envelope_json_str = msghandler.smartpack(
subject::String, subject::String,
data::AbstractArray{Tuple{String, Any, String}}; data::AbstractArray{Tuple{String, Any, String}};
broker_url::String = "nats://localhost:4222", broker_url::String = "nats://localhost:4222",
@@ -411,7 +527,7 @@ Receives and processes messages from your transport.
```julia ```julia
using msghandler using msghandler
env = msghandler.smartunpack( message_envelope = msghandler.smartunpack(
msg_json_str::String; msg_json_str::String;
fileserver_download_handler::Function = _fetch_with_backoff, fileserver_download_handler::Function = _fetch_with_backoff,
max_retries::Int = 5, max_retries::Int = 5,
@@ -438,7 +554,7 @@ data = [
("large_document", large_file_data, "binary") ("large_document", large_file_data, "binary")
] ]
env, env_json_str = smartpack("/chat/room1", data; fileserver_url="http://localhost:8080") message_envelope, message_envelope_json_str = smartpack("/chat/room1", data; fileserver_url="http://localhost:8080")
``` ```
### Example 2: Dictionary Exchange (Julia) ### Example 2: Dictionary Exchange (Julia)
@@ -455,7 +571,7 @@ config = Dict(
) )
data = [("config", config, "dictionary")] data = [("config", config, "dictionary")]
env, env_json_str = smartpack("/device/config", data) message_envelope, message_envelope_json_str = smartpack("/device/config", data)
``` ```
### Example 3: Table Data (Arrow IPC - Julia Only) ### Example 3: Table Data (Arrow IPC - Julia Only)
@@ -473,7 +589,7 @@ df = DataFrame(
) )
data = [("students", df, "arrowtable")] data = [("students", df, "arrowtable")]
env, env_json_str = smartpack("/data/analysis", data) message_envelope, message_envelope_json_str = smartpack("/data/analysis", data)
``` ```
### Example 3b: Table Data (JSON - Julia Compatible) ### Example 3b: Table Data (JSON - Julia Compatible)
@@ -491,7 +607,7 @@ df = DataFrame(
) )
data = [("students", df, "jsontable")] data = [("students", df, "jsontable")]
env, env_json_str = smartpack("/data/analysis", data) message_envelope, message_envelope_json_str = smartpack("/data/analysis", data)
``` ```
### Example 3c: Table Data (JSON - JavaScript Compatible) ### Example 3c: Table Data (JSON - JavaScript Compatible)
@@ -505,7 +621,7 @@ const tableData = [
{ id: 3, name: "Charlie", score: 92 } { id: 3, name: "Charlie", score: 92 }
]; ];
const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack("/data/analysis", [ const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack("/data/analysis", [
["students", tableData, "jsontable"] ["students", tableData, "jsontable"]
]); ]);
``` ```
@@ -522,7 +638,7 @@ image_path = "./test/large_image.png"
image_data = read(image_path) image_data = read(image_path)
data = [("user_avatar", image_data, "binary")] data = [("user_avatar", image_data, "binary")]
env, env_json_str = smartpack("/chat/room1", data; fileserver_url="http://localhost:8080") message_envelope, message_envelope_json_str = smartpack("/chat/room1", data; fileserver_url="http://localhost:8080")
``` ```
### Example 5: Image Transmission (JavaScript) ### Example 5: Image Transmission (JavaScript)
@@ -534,7 +650,7 @@ import fs from 'fs';
const file_data = fs.readFileSync('./test/large_image.png'); const file_data = fs.readFileSync('./test/large_image.png');
const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack("/chat/room1", [ const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack("/chat/room1", [
["user_avatar", file_data, "binary"] ["user_avatar", file_data, "binary"]
]); ]);
``` ```
@@ -550,7 +666,7 @@ const tableData = [
{ id: 3, name: "Charlie", score: 92 } { id: 3, name: "Charlie", score: 92 }
]; ];
const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack("/data/analysis", [ const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack("/data/analysis", [
["students", tableData, "jsontable"] ["students", tableData, "jsontable"]
]); ]);
``` ```
@@ -563,7 +679,7 @@ Bi-directional communication.
using msghandler, NATS using msghandler, NATS
# Requester # Requester
env, env_json_str = msghandler.smartpack( message_envelope, message_envelope_json_str = msghandler.smartpack(
"/device/command", "/device/command",
[("command", Dict("action" => "read_sensor"), "dictionary")]; [("command", Dict("action" => "read_sensor"), "dictionary")];
broker_url="nats://localhost:4222", broker_url="nats://localhost:4222",
@@ -571,24 +687,24 @@ env, env_json_str = msghandler.smartpack(
) )
conn = NATS.connect("nats://localhost:4222") conn = NATS.connect("nats://localhost:4222")
NATS.publish(conn, "/device/command", env_json_str) NATS.publish(conn, "/device/command", message_envelope_json_str)
NATS.drain(conn) NATS.drain(conn)
# Receiver (in separate application) # Receiver (in separate application)
conn = NATS.connect("nats://localhost:4222") conn = NATS.connect("nats://localhost:4222")
NATS.subscribe(conn, "/device/command") do msg NATS.subscribe(conn, "/device/command") do msg
env = msghandler.smartunpack(msg) message_envelope = msghandler.smartunpack(msg)
println("Received command: ", env["payloads"]) println("Received command: ", message_envelope["payloads"])
result = Dict("value" => 42) result = Dict("value" => 42)
response_env, response_json = msghandler.smartpack( response_message_envelope, response_message_envelope_json_str = msghandler.smartpack(
"/device/response", "/device/response",
[("result", result, "dictionary")], [("result", result, "dictionary")],
reply_to="/device/command", reply_to="/device/command",
reply_to_msg_id=env["msg_id"] reply_to_msg_id=message_envelope["msg_id"]
) )
NATS.publish(conn, "/device/response", response_json) NATS.publish(conn, "/device/response", response_message_envelope_json_str)
NATS.drain(conn) NATS.drain(conn)
end end
``` ```
@@ -648,10 +764,10 @@ cat > test-browser.html << 'EOF'
import msghandlerCSR from './src/msghandler-csr.js'; import msghandlerCSR from './src/msghandler-csr.js';
// Test smartpack // Test smartpack
const [env, json] = await msghandlerCSR.smartpack("/test", [ const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack("/test", [
["msg", "Hello Browser!", "text"] ["msg", "Hello Browser!", "text"]
]); ]);
console.log("Sent:", json); console.log("Sent:", message_envelope_json_str);
</script> </script>
</body> </body>
</html> </html>
@@ -679,7 +795,7 @@ The JavaScript version provides the same core functionality as Julia, with brows
#### smartpack #### smartpack
```javascript ```javascript
const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack( const [message_envelope, message_envelope_json_str] = await msghandlerCSR.smartpack(
subject, subject,
data, // [[dataname, data, type], ...] data, // [[dataname, data, type], ...]
options options
@@ -689,8 +805,8 @@ const [envelope, envelopeJsonStr] = await msghandlerCSR.smartpack(
#### smartunpack #### smartunpack
```javascript ```javascript
const envelope = await msghandlerCSR.smartunpack(msg, options); const message_envelope = await msghandlerCSR.smartunpack(msg, options);
// envelope.payloads = [[dataname, data, type], ...] // message_envelope.payloads = [[dataname, data, type], ...]
``` ```
#### plikOneshotUpload #### plikOneshotUpload