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 19:18:12 +07:00
parent f8a92a45a0
commit 46fdf668c6
22 changed files with 260 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

189
src/NATSBridge.jl
View File

@@ -286,35 +286,35 @@ function envelope_to_json(env::msgEnvelope_v1)
# Convert payloads to JSON array
if !isempty(env.payloads)
payloads_json = []
for payload in env.payloads
payload_obj = Dict{String, Any}(
"id" => payload.id,
"dataname" => payload.dataname,
"type" => payload.type,
"transport" => payload.transport,
"encoding" => payload.encoding,
"size" => payload.size,
)
# Include data based on transport type
if payload.transport == "direct" && payload.data !== nothing
if payload.encoding == "base64" || payload.encoding == "json"
payload_obj["data"] = payload.data
else
# For other encodings, use base64
payload_bytes = _get_payload_bytes(payload.data)
payload_obj["data"] = Base64.base64encode(payload_bytes)
end
elseif payload.transport == "link" && payload.data !== nothing
# For link transport, data is a URL string - include directly
payload_obj["data"] = payload.data
end
if !isempty(payload.metadata)
payload_obj["metadata"] = Dict(String(k) => v for (k, v) in payload.metadata)
payloads_json = []
for payload in env.payloads
payload_obj = Dict{String, Any}(
"id" => payload.id,
"dataname" => payload.dataname,
"type" => payload.type,
"transport" => payload.transport,
"encoding" => payload.encoding,
"size" => payload.size,
)
# Include data based on transport type
if payload.transport == "direct" && payload.data !== nothing
if payload.encoding == "base64" || payload.encoding == "json"
payload_obj["data"] = payload.data
else
# For other encodings, use base64
payload_bytes = _get_payload_bytes(payload.data)
payload_obj["data"] = Base64.base64encode(payload_bytes)
end
push!(payloads_json, payload_obj)
elseif payload.transport == "link" && payload.data !== nothing
# For link transport, data is a URL string - include directly
payload_obj["data"] = payload.data
end
if !isempty(payload.metadata)
payload_obj["metadata"] = Dict(String(k) => v for (k, v) in payload.metadata)
end
obj["payloads"] = payloads_json
push!(payloads_json, payload_obj)
end
obj["payloads"] = payloads_json
end
JSON.json(obj)
@@ -361,6 +361,7 @@ Each payload can have a different type, enabling mixed-content messages (e.g., c
3. Compares the serialized size against `size_threshold`
4. For small payloads: encodes as Base64, constructs a "direct" msgPayload_v1
5. For large payloads: uploads to the fileserver, constructs a "link" msgPayload_v1 with the URL
6. Converts envelope to JSON string and optionally publishes to NATS
# Arguments:
- `subject::String` - NATS subject to publish the message to
@@ -372,18 +373,22 @@ Each payload can have a different type, enabling mixed-content messages (e.g., c
# Keyword Arguments:
- `nats_url::String = DEFAULT_NATS_URL` - URL of the NATS server
- `fileserver_url = DEFAULT_FILESERVER_URL` - URL of the HTTP file server for large payloads
- `fileserverUploadHandler::Function = plik_oneshot_upload` - Function to handle fileserver uploads (must return Dict with "status", "uploadid", "fileid", "url" keys)
- `size_threshold::Int = DEFAULT_SIZE_THRESHOLD` - Threshold in bytes separating direct vs link transport
- `correlation_id::Union{String, Nothing} = nothing` - Optional correlation ID for tracing; if `nothing`, a UUID is generated
- `msg_purpose::String = "chat"` - Purpose of the message: "ACK", "NACK", "updateStatus", "shutdown", "chat", etc.
- `sender_name::String = "NATSBridge"` - Name of the sender
- `sender_name::String = "default"` - Name of the sender
- `receiver_name::String = ""` - Name of the receiver (empty string means broadcast)
- `receiver_id::String = ""` - UUID of the receiver (empty string means broadcast)
- `reply_to::String = ""` - Topic to reply to (empty string if no reply expected)
- `reply_to_msg_id::String = ""` - Message ID this message is replying to
- `is_publish::Bool = true` - Whether to automatically publish the message to NATS
# Return:
- A `msgEnvelope_v1` object containing metadata and transport information
- A tuple `(env, msg_json_str)` where:
- `env::msgEnvelope_v1` - The envelope object containing all metadata and payloads
- `msg_json_str::String` - JSON string representation of the envelope for publishing
# Example
```jldoctest
@@ -391,39 +396,43 @@ using UUIDs
# Send a single payload (still wrapped in a list)
data = Dict("key" => "value")
env = smartsend("my.subject", [("dataname1", data, "dictionary")])
env, msg_json = smartsend("my.subject", [("dataname1", data, "dictionary")])
# Send multiple payloads in one message with different types
data1 = Dict("key1" => "value1")
data2 = rand(10_000) # Small array
env = smartsend("my.subject", [("dataname1", data1, "dictionary"), ("dataname2", data2, "table")])
env, msg_json = smartsend("my.subject", [("dataname1", data1, "dictionary"), ("dataname2", data2, "table")])
# Send a large array using fileserver upload
data = rand(10_000_000) # ~80 MB
env = smartsend("large.data", [("large_table", data, "table")])
env, msg_json = smartsend("large.data", [("large_table", data, "table")])
# Mixed content (e.g., chat with text and image)
env = smartsend("chat.subject", [
env, msg_json = smartsend("chat.subject", [
("message_text", "Hello!", "text"),
("user_image", image_data, "image"),
("audio_clip", audio_data, "audio")
])
# Publish the JSON string directly using NATS request-reply pattern
# reply = NATS.request(nats_url, subject, msg_json_str; reply_to=reply_to_topic)
```
"""
""" #[PENDING]
function smartsend(
subject::String, # smartreceive's subject
data::AbstractArray{Tuple{String, T1, String}, 1}; # List of (dataname, data, type) tuples
data::AbstractArray{Tuple{String, T1, String}, 1}; # List of (dataname, data, type) tuples. Use Tuple{String, Any, String}[] for empty payloads
nats_url::String = DEFAULT_NATS_URL,
fileserver_url = DEFAULT_FILESERVER_URL,
fileserverUploadHandler::Function=plik_oneshot_upload, # a function to handle uploading data to specific HTTP fileserver
size_threshold::Int = DEFAULT_SIZE_THRESHOLD,
correlation_id::Union{String, Nothing} = nothing,
msg_purpose::String = "chat",
sender_name::String = "NATSBridge",
sender_name::String = "default",
receiver_name::String = "",
receiver_id::String = "",
reply_to::String = "",
reply_to_msg_id::String = ""
reply_to_msg_id::String = "",
is_publish::Bool = true # some time the user want to get env and msg_json_str from this function without publishing the msg
) where {T1<:Any}
# Generate correlation ID if not provided
@@ -437,59 +446,59 @@ function smartsend(
# Process each payload in the list
payloads = msgPayload_v1[]
for (dataname, payload_data, payload_type) in data
# Serialize data based on type
payload_bytes = _serialize_data(payload_data, payload_type)
# Serialize data based on type
payload_bytes = _serialize_data(payload_data, payload_type)
payload_size = length(payload_bytes) # Calculate payload size in bytes
log_trace(cid, "Serialized payload '$dataname' (type: $payload_type) size: $payload_size bytes") # Log payload size
# Decision: Direct vs Link
if payload_size < size_threshold # Check if payload is small enough for direct transport
# Direct path - Base64 encode and send via NATS
payload_b64 = Base64.base64encode(payload_bytes) # Encode bytes as base64 string
log_trace(cid, "Using direct transport for $payload_size bytes") # Log transport choice
payload_size = length(payload_bytes) # Calculate payload size in bytes
log_trace(cid, "Serialized payload '$dataname' (type: $payload_type) size: $payload_size bytes") # Log payload size
# Create msgPayload_v1 for direct transport
payload = msgPayload_v1(
payload_b64,
payload_type;
id = string(uuid4()),
dataname = dataname,
transport = "direct",
encoding = "base64",
size = payload_size,
metadata = Dict{String, Any}("payload_bytes" => payload_size)
)
push!(payloads, payload)
else
# Link path - Upload to HTTP server, send URL via NATS
log_trace(cid, "Using link transport, uploading to fileserver") # Log link transport choice
# Decision: Direct vs Link
if payload_size < size_threshold # Check if payload is small enough for direct transport
# Direct path - Base64 encode and send via NATS
payload_b64 = Base64.base64encode(payload_bytes) # Encode bytes as base64 string
log_trace(cid, "Using direct transport for $payload_size bytes") # Log transport choice
# Create msgPayload_v1 for direct transport
payload = msgPayload_v1(
payload_b64,
payload_type;
id = string(uuid4()),
dataname = dataname,
transport = "direct",
encoding = "base64",
size = payload_size,
metadata = Dict{String, Any}("payload_bytes" => payload_size)
)
push!(payloads, payload)
else
# Link path - Upload to HTTP server, send URL via NATS
log_trace(cid, "Using link transport, uploading to fileserver") # Log link transport choice
# Upload to HTTP server
response = fileserverUploadHandler(fileserver_url, dataname, payload_bytes)
if response["status"] != 200 # Check if upload was successful
error("Failed to upload data to fileserver: $(response["status"])") # Throw error if upload failed
end
url = response["url"] # URL for the uploaded data
log_trace(cid, "Uploaded to URL: $url") # Log successful upload
# Create msgPayload_v1 for link transport
payload = msgPayload_v1(
url,
payload_type;
id = string(uuid4()),
dataname = dataname,
transport = "link",
encoding = "none",
size = payload_size,
metadata = Dict{String, Any}()
)
push!(payloads, payload)
# Upload to HTTP server
response = fileserverUploadHandler(fileserver_url, dataname, payload_bytes)
if response["status"] != 200 # Check if upload was successful
error("Failed to upload data to fileserver: $(response["status"])") # Throw error if upload failed
end
url = response["url"] # URL for the uploaded data
log_trace(cid, "Uploaded to URL: $url") # Log successful upload
# Create msgPayload_v1 for link transport
payload = msgPayload_v1(
url,
payload_type;
id = string(uuid4()),
dataname = dataname,
transport = "link",
encoding = "none",
size = payload_size,
metadata = Dict{String, Any}()
)
push!(payloads, payload)
end
end
# Create msgEnvelope_v1 with all payloads
env = msgEnvelope_v1(
subject,
@@ -507,10 +516,12 @@ function smartsend(
metadata = Dict{String, Any}(),
)
msg_json = envelope_to_json(env) # Convert envelope to JSON
publish_message(nats_url, subject, msg_json, cid) # Publish message to NATS
msg_json_str = envelope_to_json(env) # Convert envelope to JSON
if is_publish
publish_message(nats_url, subject, msg_json_str, cid) # Publish message to NATS
end
return env # Return the envelope for tracking
return (env, msg_json_str)
end