ton/NATSBridge
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-02-23 22:00:06 +07:00
parent 7853e94d2e
commit 2c340e37c7
1 changed files with 45 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

90
README.md
View File

@@ -173,7 +173,7 @@ from nats_bridge import smartsend
# Send a text message # Send a text message
data = [("message", "Hello World", "text")] data = [("message", "Hello World", "text")]
envelope, msg_json_str = smartsend("/chat/room1", data, nats_url="nats://localhost:4222") env, env_json_str = smartsend("/chat/room1", data, nats_url="nats://localhost:4222")
print("Message sent!") print("Message sent!")
``` ```
@@ -183,7 +183,7 @@ print("Message sent!")
const { smartsend } = require('./src/NATSBridge'); const { smartsend } = require('./src/NATSBridge');
// Send a text message // Send a text message
const { envelope, msg_json_str } = await smartsend("/chat/room1", [ const { env, env_json_str } = await smartsend("/chat/room1", [
{ dataname: "message", data: "Hello World", type: "text" } { dataname: "message", data: "Hello World", type: "text" }
], { natsUrl: "nats://localhost:4222" }); ], { natsUrl: "nats://localhost:4222" });
@@ -197,7 +197,7 @@ using NATSBridge
# Send a text message # Send a text message
data = [("message", "Hello World", "text")] data = [("message", "Hello World", "text")]
envelope, msg_json_str = NATSBridge.smartsend("/chat/room1", data; nats_url="nats://localhost:4222") env, env_json_str = NATSBridge.smartsend("/chat/room1", data; nats_url="nats://localhost:4222")
println("Message sent!") println("Message sent!")
``` ```
@@ -221,8 +221,8 @@ async def main():
# Subscribe to the subject - msg comes from the callback # Subscribe to the subject - msg comes from the callback
async def message_handler(msg): async def message_handler(msg):
# Receive and process message # Receive and process message
envelope = smartreceive(msg.data) env = smartreceive(msg.data)
for dataname, data, type in envelope["payloads"]: for dataname, data, type in env["payloads"]:
print(f"Received {dataname}: {data}") print(f"Received {dataname}: {data}")
sid = await nc.subscribe(SUBJECT, cb=message_handler) sid = await nc.subscribe(SUBJECT, cb=message_handler)
@@ -251,8 +251,8 @@ async function main() {
for await (const msg of sub) { for await (const msg of sub) {
// Receive and process message // Receive and process message
const envelope = await smartreceive(msg); const env = await smartreceive(msg);
for (const payload of envelope.payloads) { for (const payload of env.payloads) {
console.log(`Received ${payload.dataname}: ${payload.data}`); console.log(`Received ${payload.dataname}: ${payload.data}`);
} }
} }
@@ -283,8 +283,8 @@ function test_receive()
log_trace("Received message on $(msg.subject)") log_trace("Received message on $(msg.subject)")
# Receive and process message # Receive and process message
envelope, msg_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler) env, env_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler)
for (dataname, data, type) in envelope["payloads"] for (dataname, data, type) in env["payloads"]
println("Received $dataname: $data") println("Received $dataname: $data")
end end
end end
@@ -310,7 +310,7 @@ Sends data either directly via NATS or via a fileserver URL, depending on payloa
```python ```python
from nats_bridge import smartsend from nats_bridge import smartsend
envelope, msg_json_str = smartsend( env, env_json_str = smartsend(
subject, # NATS subject to publish to subject, # NATS subject to publish to
data, # List of (dataname, data, type) tuples data, # List of (dataname, data, type) tuples
nats_url="nats://localhost:4222", # NATS server URL nats_url="nats://localhost:4222", # NATS server URL
@@ -333,7 +333,7 @@ envelope, msg_json_str = smartsend(
```javascript ```javascript
const { smartsend } = require('./src/NATSBridge'); const { smartsend } = require('./src/NATSBridge');
const { envelope, msg_json_str } = await smartsend( const { env, env_json_str } = await smartsend(
subject, // NATS subject subject, // NATS subject
data, // Array of {dataname, data, type} data, // Array of {dataname, data, type}
{ {
@@ -358,7 +358,7 @@ const { envelope, msg_json_str } = await smartsend(
```julia ```julia
using NATSBridge using NATSBridge
envelope, msg_json_str = NATSBridge.smartsend( env, env_json_str = NATSBridge.smartsend(
subject, # NATS subject subject, # NATS subject
data::AbstractArray{Tuple{String, Any, String}}; # List of (dataname, data, type) data::AbstractArray{Tuple{String, Any, String}}; # List of (dataname, data, type)
nats_url::String = "nats://localhost:4222", nats_url::String = "nats://localhost:4222",
@@ -375,8 +375,8 @@ envelope, msg_json_str = NATSBridge.smartsend(
is_publish::Bool = true # Whether to automatically publish to NATS is_publish::Bool = true # Whether to automatically publish to NATS
) )
# Returns: (msgEnvelope_v1, JSON string) # Returns: (msgEnvelope_v1, JSON string)
# - envelope: msgEnvelope_v1 object with all envelope metadata and payloads # - env: msgEnvelope_v1 object with all envelope metadata and payloads
# - msg_json_str: JSON string representation of the envelope for publishing # - env_json_str: JSON string representation of the envelope for publishing
``` ```
### smartreceive ### smartreceive
@@ -389,7 +389,7 @@ Receives and processes messages from NATS, handling both direct and link transpo
from nats_bridge import smartreceive from nats_bridge import smartreceive
# Note: For nats-py, use msg.data to pass the raw message data # Note: For nats-py, use msg.data to pass the raw message data
envelope = smartreceive( env = smartreceive(
msg.data, # NATS message data (msg.data for nats-py) msg.data, # NATS message data (msg.data for nats-py)
fileserver_download_handler=_fetch_with_backoff, # Download handler fileserver_download_handler=_fetch_with_backoff, # Download handler
max_retries=5, # Max retry attempts max_retries=5, # Max retry attempts
@@ -405,7 +405,7 @@ envelope = smartreceive(
const { smartreceive } = require('./src/NATSBridge'); const { smartreceive } = require('./src/NATSBridge');
// Note: msg is the NATS message object from subscription // Note: msg is the NATS message object from subscription
const envelope = await smartreceive( const env = await smartreceive(
msg, // NATS message (raw object from subscription) msg, // NATS message (raw object from subscription)
{ {
fileserverDownloadHandler: customDownloadHandler, fileserverDownloadHandler: customDownloadHandler,
@@ -423,7 +423,7 @@ const envelope = await smartreceive(
using NATSBridge using NATSBridge
# Note: msg is a NATS.Msg object passed from the subscription callback # Note: msg is a NATS.Msg object passed from the subscription callback
envelope, msg_json_str = NATSBridge.smartreceive( env, env_json_str = NATSBridge.smartreceive(
msg::NATS.Msg; msg::NATS.Msg;
fileserverDownloadHandler::Function = _fetch_with_backoff, fileserverDownloadHandler::Function = _fetch_with_backoff,
max_retries::Int = 5, max_retries::Int = 5,
@@ -517,14 +517,14 @@ data = [
("large_document", large_file_data, "binary") ("large_document", large_file_data, "binary")
] ]
envelope, msg_json_str = smartsend("/chat/room1", data, fileserver_url="http://localhost:8080") env, env_json_str = smartsend("/chat/room1", data, fileserver_url="http://localhost:8080")
``` ```
#### JavaScript #### JavaScript
```javascript ```javascript
const { smartsend } = require('./src/NATSBridge'); const { smartsend } = require('./src/NATSBridge');
const { envelope, msg_json_str } = await smartsend("/chat/room1", [ const { env, env_json_str } = await smartsend("/chat/room1", [
{ dataname: "message_text", data: "Hello!", type: "text" }, { dataname: "message_text", data: "Hello!", type: "text" },
{ dataname: "user_avatar", data: image_data, type: "image" }, { dataname: "user_avatar", data: image_data, type: "image" },
{ dataname: "large_document", data: large_file_data, type: "binary" } { dataname: "large_document", data: large_file_data, type: "binary" }
@@ -543,7 +543,7 @@ data = [
("large_document", large_file_data, "binary") ("large_document", large_file_data, "binary")
] ]
envelope, msg_json_str = NATSBridge.smartsend("/chat/room1", data; fileserver_url="http://localhost:8080") env, env_json_str = NATSBridge.smartsend("/chat/room1", data; fileserver_url="http://localhost:8080")
``` ```
### Example 2: Dictionary Exchange ### Example 2: Dictionary Exchange
@@ -561,7 +561,7 @@ config = {
} }
data = [("config", config, "dictionary")] data = [("config", config, "dictionary")]
envelope, msg_json_str = smartsend("/device/config", data) env, env_json_str = smartsend("/device/config", data)
``` ```
#### JavaScript #### JavaScript
@@ -574,7 +574,7 @@ const config = {
update_interval: 60 update_interval: 60
}; };
const { envelope, msg_json_str } = await smartsend("/device/config", [ const { env, env_json_str } = await smartsend("/device/config", [
{ dataname: "config", data: config, type: "dictionary" } { dataname: "config", data: config, type: "dictionary" }
]); ]);
``` ```
@@ -590,7 +590,7 @@ config = Dict(
) )
data = [("config", config, "dictionary")] data = [("config", config, "dictionary")]
envelope, msg_json_str = NATSBridge.smartsend("/device/config", data) env, env_json_str = NATSBridge.smartsend("/device/config", data)
``` ```
### Example 3: Table Data (Arrow IPC) ### Example 3: Table Data (Arrow IPC)
@@ -609,7 +609,7 @@ df = pd.DataFrame({
}) })
data = [("students", df, "table")] data = [("students", df, "table")]
envelope, msg_json_str = smartsend("/data/analysis", data) env, env_json_str = smartsend("/data/analysis", data)
``` ```
#### JavaScript #### JavaScript
@@ -622,7 +622,7 @@ const tableData = [
{ id: 3, name: "Charlie", score: 92 } { id: 3, name: "Charlie", score: 92 }
]; ];
const { envelope, msg_json_str } = await smartsend("/data/analysis", [ const { env, env_json_str } = await smartsend("/data/analysis", [
{ dataname: "students", data: tableData, type: "table" } { dataname: "students", data: tableData, type: "table" }
]); ]);
``` ```
@@ -639,7 +639,7 @@ df = DataFrame(
) )
data = [("students", df, "table")] data = [("students", df, "table")]
envelope, msg_json_str = NATSBridge.smartsend("/data/analysis", data) env, env_json_str = NATSBridge.smartsend("/data/analysis", data)
``` ```
### Example 4: Request-Response Pattern with Envelope JSON ### Example 4: Request-Response Pattern with Envelope JSON
@@ -650,16 +650,16 @@ Bi-directional communication with reply-to support. The `smartsend` function now
```python ```python
from nats_bridge import smartsend from nats_bridge import smartsend
envelope, msg_json_str = smartsend( env, env_json_str = smartsend(
"/device/command", "/device/command",
[("command", {"action": "read_sensor"}, "dictionary")], [("command", {"action": "read_sensor"}, "dictionary")],
reply_to="/device/response" reply_to="/device/response"
) )
# envelope: msgEnvelope_v1 object # env: msgEnvelope_v1 object
# msg_json_str: JSON string for publishing to NATS # env_json_str: JSON string for publishing to NATS
# The msg_json_str can also be published directly using NATS request-reply pattern # The env_json_str can also be published directly using NATS request-reply pattern
# nc.request("/device/command", msg_json_str, reply_to="/device/response") # nc.request("/device/command", env_json_str, reply_to="/device/response")
``` ```
#### Python/Micropython (Responder) #### Python/Micropython (Responder)
@@ -677,8 +677,8 @@ async def main():
nc = await nats.connect(NATS_URL) nc = await nats.connect(NATS_URL)
async def message_handler(msg): async def message_handler(msg):
envelope = smartreceive(msg.data) env = smartreceive(msg.data)
for dataname, data, type in envelope["payloads"]: for dataname, data, type in env["payloads"]:
if data.get("action") == "read_sensor": if data.get("action") == "read_sensor":
response = {"sensor_id": "sensor-001", "value": 42.5} response = {"sensor_id": "sensor-001", "value": 42.5}
smartsend(REPLY_SUBJECT, [("data", response, "dictionary")]) smartsend(REPLY_SUBJECT, [("data", response, "dictionary")])
@@ -694,7 +694,7 @@ asyncio.run(main())
```javascript ```javascript
const { smartsend } = require('./src/NATSBridge'); const { smartsend } = require('./src/NATSBridge');
const { envelope, msg_json_str } = await smartsend("/device/command", [ const { env, env_json_str } = await smartsend("/device/command", [
{ dataname: "command", data: { action: "read_sensor" }, type: "dictionary" } { dataname: "command", data: { action: "read_sensor" }, type: "dictionary" }
], { ], {
replyTo: "/device/response" replyTo: "/device/response"
@@ -717,8 +717,8 @@ async function main() {
const sub = nc.subscribe(SUBJECT); const sub = nc.subscribe(SUBJECT);
for await (const msg of sub) { for await (const msg of sub) {
const envelope = await smartreceive(msg); const env = await smartreceive(msg);
for (const payload of envelope.payloads) { for (const payload of env.payloads) {
if (payload.dataname === "command" && payload.data.action === "read_sensor") { if (payload.dataname === "command" && payload.data.action === "read_sensor") {
const response = { sensor_id: "sensor-001", value: 42.5 }; const response = { sensor_id: "sensor-001", value: 42.5 };
await smartsend(REPLY_SUBJECT, [ await smartsend(REPLY_SUBJECT, [
@@ -736,7 +736,7 @@ main();
```julia ```julia
using NATSBridge using NATSBridge
envelope, msg_json_str = NATSBridge.smartsend( env, env_json_str = NATSBridge.smartsend(
"/device/command", "/device/command",
[("command", Dict("action" => "read_sensor"), "dictionary")]; [("command", Dict("action" => "read_sensor"), "dictionary")];
reply_to="/device/response" reply_to="/device/response"
@@ -755,8 +755,8 @@ const NATS_URL = "nats://localhost:4222"
function test_responder() function test_responder()
conn = NATS.connect(NATS_URL) conn = NATS.connect(NATS_URL)
NATS.subscribe(conn, SUBJECT) do msg NATS.subscribe(conn, SUBJECT) do msg
envelope, msg_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler) env, env_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler)
for (dataname, data, type) in envelope["payloads"] for (dataname, data, type) in env["payloads"]
if dataname == "command" && data["action"] == "read_sensor" if dataname == "command" && data["action"] == "read_sensor"
response = Dict("sensor_id" => "sensor-001", "value" => 42.5) response = Dict("sensor_id" => "sensor-001", "value" => 42.5)
smartsend(REPLY_SUBJECT, [("data", response, "dictionary")]) smartsend(REPLY_SUBJECT, [("data", response, "dictionary")])
@@ -794,8 +794,8 @@ async def main():
# Receive commands - msg comes from the callback # Receive commands - msg comes from the callback
async def message_handler(msg): async def message_handler(msg):
envelope = smartreceive(msg.data) env = smartreceive(msg.data)
for dataname, data, type in envelope["payloads"]: for dataname, data, type in env["payloads"]:
if type == "dictionary" and data.get("action") == "reboot": if type == "dictionary" and data.get("action") == "reboot":
# Execute reboot # Execute reboot
pass pass
@@ -822,8 +822,8 @@ async function main() {
const sub = nc.subscribe(SUBJECT); const sub = nc.subscribe(SUBJECT);
for await (const msg of sub) { for await (const msg of sub) {
const envelope = await smartreceive(msg); const env = await smartreceive(msg);
for (const payload of envelope.payloads) { for (const payload of env.payloads) {
if (payload.dataname === "temperature") { if (payload.dataname === "temperature") {
console.log(`Temperature: ${payload.data}`); console.log(`Temperature: ${payload.data}`);
} else if (payload.dataname === "humidity") { } else if (payload.dataname === "humidity") {
@@ -847,8 +847,8 @@ const NATS_URL = "nats://localhost:4222"
function test_receiver() function test_receiver()
conn = NATS.connect(NATS_URL) conn = NATS.connect(NATS_URL)
NATS.subscribe(conn, SUBJECT) do msg NATS.subscribe(conn, SUBJECT) do msg
envelope, msg_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler) env, env_json_str = NATSBridge.smartreceive(msg, fileserverDownloadHandler)
for (dataname, data, type) in envelope["payloads"] for (dataname, data, type) in env["payloads"]
if dataname == "temperature" if dataname == "temperature"
println("Temperature: $data") println("Temperature: $data")
elseif dataname == "humidity" elseif dataname == "humidity"