update
This commit is contained in:
+27
-79
@@ -416,15 +416,15 @@ env, msg_json = smartpack("chat.subject", [
|
|||||||
|
|
||||||
# Custom fileserver upload handler (e.g., for AWS S3)
|
# Custom fileserver upload handler (e.g., for AWS S3)
|
||||||
function my_custom_upload(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any}
|
function my_custom_upload(fileserver_url::String, dataname::String, data::Vector{UInt8})::Dict{String, Any}
|
||||||
# Upload `data` to your file server at `fileserver_url`, using `dataname` as the file name
|
# Upload `data` to your file server at `fileserver_url`, using `dataname` as the file name
|
||||||
# Return a Dict with required keys:
|
# Return a Dict with required keys:
|
||||||
# "status" - HTTP status code (e.g. 200)
|
# "status" - HTTP status code (e.g. 200)
|
||||||
# "uploadid" - String ID of the upload session
|
# "uploadid" - String ID of the upload session
|
||||||
# "fileid" - String ID of the uploaded file
|
# "fileid" - String ID of the uploaded file
|
||||||
# "url" - Full download URL for the uploaded file
|
# "url" - Full download URL for the uploaded file
|
||||||
# Example return:
|
# Example return:
|
||||||
# Dict("status" => 200, "uploadid" => "abc123", "fileid" => "file456", "url" => "http://my-server.com/abc123/file456/myfile.bin")
|
# Dict("status" => 200, "uploadid" => "abc123", "fileid" => "file456", "url" => "http://my-server.com/abc123/file456/myfile.bin")
|
||||||
throw(NotImplementedError("implement upload logic here"))
|
throw(NotImplementedError("implement upload logic here"))
|
||||||
end
|
end
|
||||||
```
|
```
|
||||||
"""
|
"""
|
||||||
@@ -723,75 +723,6 @@ function _serialize_data(data::Any, payload_type::String)
|
|||||||
end
|
end
|
||||||
|
|
||||||
|
|
||||||
# """ publish_message - Publish message via transport
|
|
||||||
# This function publishes a message via the transport with proper
|
|
||||||
# connection management and logging.
|
|
||||||
|
|
||||||
# # Arguments:
|
|
||||||
# - `broker_url::String` - Broker URL (e.g., "localhost:4222")
|
|
||||||
# - `subject::String` - Subject to publish to (e.g., "/agent/wine/api/v1/prompt")
|
|
||||||
# - `message::String` - JSON message to publish
|
|
||||||
# - `correlation_id::String` - Correlation ID for tracing and logging
|
|
||||||
|
|
||||||
# # Return:
|
|
||||||
# - `nothing` - This function performs publishing but returns nothing
|
|
||||||
|
|
||||||
# # Example
|
|
||||||
# ```jldoctest
|
|
||||||
# using NATS
|
|
||||||
|
|
||||||
# # Prepare JSON message
|
|
||||||
# message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
|
|
||||||
|
|
||||||
# # Publish via transport
|
|
||||||
# publish_message("localhost:4222", "my.subject", message, "abc123")
|
|
||||||
# ```
|
|
||||||
# """
|
|
||||||
# function publish_message(broker_url::String, subject::String, message::String, correlation_id::String)
|
|
||||||
# conn = NATS.connect(broker_url) # Create NATS connection
|
|
||||||
# publish_message(conn, subject, message, correlation_id)
|
|
||||||
# end
|
|
||||||
|
|
||||||
# """ publish_message - Publish message via transport using pre-existing connection
|
|
||||||
# This function publishes a message via the transport using a pre-existing connection,
|
|
||||||
# avoiding the overhead of connection establishment.
|
|
||||||
|
|
||||||
# # Arguments:
|
|
||||||
# - `conn` - Pre-existing connection object with publish/close methods
|
|
||||||
# - `subject::String` - Subject to publish to (e.g., "/agent/wine/api/v1/prompt")
|
|
||||||
# - `message::String` - JSON message to publish
|
|
||||||
# - `correlation_id::String` - Correlation ID for tracing and logging
|
|
||||||
|
|
||||||
# # Return:
|
|
||||||
# - `nothing` - This function performs publishing but returns nothing
|
|
||||||
|
|
||||||
# # Example
|
|
||||||
# ```jldoctest
|
|
||||||
# using NATS
|
|
||||||
|
|
||||||
# # Prepare JSON message
|
|
||||||
# message = "{\"correlation_id\":\"abc123\",\"payload\":\"test\"}"
|
|
||||||
|
|
||||||
# # Create connection once and reuse for multiple publishes
|
|
||||||
# conn = NATS.connect("nats://localhost:4222")
|
|
||||||
# publish_message(conn, "my.subject", message, "abc123")
|
|
||||||
# # Connection is automatically drained after publish
|
|
||||||
# ```
|
|
||||||
|
|
||||||
# # Use Case:
|
|
||||||
# Use this version when you already have an established connection and want to publish
|
|
||||||
# multiple messages without the overhead of creating a new connection for each publish.
|
|
||||||
# """
|
|
||||||
# function publish_message(conn::NATS.Connection, subject::String, message::String, correlation_id::String)
|
|
||||||
# try
|
|
||||||
# NATS.publish(conn, subject, message) # Publish message to NATS
|
|
||||||
# log_trace(correlation_id, "Message published to $subject") # Log successful publish
|
|
||||||
# finally
|
|
||||||
# NATS.drain(conn) # Ensure connection is closed properly
|
|
||||||
# end
|
|
||||||
# end
|
|
||||||
|
|
||||||
|
|
||||||
""" smartunpack - Receive and process messages
|
""" smartunpack - Receive and process messages
|
||||||
This function processes incoming messages, handling both direct transport
|
This function processes incoming messages, handling both direct transport
|
||||||
(base64 decoded payloads) and link transport (URL-based payloads).
|
(base64 decoded payloads) and link transport (URL-based payloads).
|
||||||
@@ -809,7 +740,7 @@ A HTTP file server is required along with its download function.
|
|||||||
- `msg_json_str::String` - JSON string from the message payload (e.g., `String(msg.payload)`)
|
- `msg_json_str::String` - JSON string from the message payload (e.g., `String(msg.payload)`)
|
||||||
|
|
||||||
# Keyword Arguments:
|
# Keyword Arguments:
|
||||||
- `fileserver_download_handler::Function = _fetch_with_backoff` - Function to handle downloading data from file server URLs
|
- `fileserver_download_handler::Function = _fetch_with_backoff` - Custom download handler with signature `handler(url::String, max_retries::Int, base_delay::Int, max_delay::Int, correlation_id::String)::Vector{UInt8}`. Must accept the download URL, retry/backoff parameters, and correlation ID; return the fetched data as `Vector{UInt8}` bytes.
|
||||||
- `max_retries::Int = 5` - Maximum retry attempts for fetching URL
|
- `max_retries::Int = 5` - Maximum retry attempts for fetching URL
|
||||||
- `base_delay::Int = 100` - Initial delay for exponential backoff in ms
|
- `base_delay::Int = 100` - Initial delay for exponential backoff in ms
|
||||||
- `max_delay::Int = 5000` - Maximum delay for exponential backoff in ms
|
- `max_delay::Int = 5000` - Maximum delay for exponential backoff in ms
|
||||||
@@ -823,6 +754,23 @@ A HTTP file server is required along with its download function.
|
|||||||
msg_json_str = String(msg.payload)
|
msg_json_str = String(msg.payload)
|
||||||
env = smartunpack(msg_json_str; fileserver_download_handler=_fetch_with_backoff, max_retries=5, base_delay=100, max_delay=5000)
|
env = smartunpack(msg_json_str; fileserver_download_handler=_fetch_with_backoff, max_retries=5, base_delay=100, max_delay=5000)
|
||||||
# env["payloads"] = [("dataname1", data1, "type1"), ("dataname2", data2, "type2"), ...]
|
# env["payloads"] = [("dataname1", data1, "type1"), ("dataname2", data2, "type2"), ...]
|
||||||
|
|
||||||
|
# Custom fileserver download handler (e.g., for AWS S3 presigned URLs)
|
||||||
|
function my_custom_download(url::String, max_retries::Int, base_delay::Int, max_delay::Int, correlation_id::String)::Vector{UInt8}
|
||||||
|
# Fetch data from `url` with retry logic using exponential backoff
|
||||||
|
# Return the response body as Vector{UInt8} bytes
|
||||||
|
# Example:
|
||||||
|
# for attempt in 1:max_retries
|
||||||
|
# resp = HTTP.request("GET", url)
|
||||||
|
# if resp.status == 200
|
||||||
|
# return resp.body
|
||||||
|
# end
|
||||||
|
# end
|
||||||
|
throw(NotImplementedError("implement download logic here"))
|
||||||
|
end
|
||||||
|
|
||||||
|
# Use custom download handler
|
||||||
|
env = smartunpack(msg_json_str; fileserver_download_handler=my_custom_download)
|
||||||
```
|
```
|
||||||
"""
|
"""
|
||||||
function smartunpack(
|
function smartunpack(
|
||||||
|
|||||||
Reference in New Issue
Block a user