From 272c021665569ee47a029d837109d404f889428c Mon Sep 17 00:00:00 2001 From: narawat Date: Fri, 28 Aug 2026 17:34:33 +0700 Subject: [PATCH] update --- src/msghandler.jl | 106 ++++++++++++---------------------------------- 1 file changed, 27 insertions(+), 79 deletions(-) diff --git a/src/msghandler.jl b/src/msghandler.jl index cafa90b..9f91b84 100644 --- a/src/msghandler.jl +++ b/src/msghandler.jl @@ -416,15 +416,15 @@ env, msg_json = smartpack("chat.subject", [ # Custom fileserver upload handler (e.g., for AWS S3) 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 - # Return a Dict with required keys: - # "status" - HTTP status code (e.g. 200) - # "uploadid" - String ID of the upload session - # "fileid" - String ID of the uploaded file - # "url" - Full download URL for the uploaded file - # Example return: - # Dict("status" => 200, "uploadid" => "abc123", "fileid" => "file456", "url" => "http://my-server.com/abc123/file456/myfile.bin") - throw(NotImplementedError("implement upload logic here")) + # Upload `data` to your file server at `fileserver_url`, using `dataname` as the file name + # Return a Dict with required keys: + # "status" - HTTP status code (e.g. 200) + # "uploadid" - String ID of the upload session + # "fileid" - String ID of the uploaded file + # "url" - Full download URL for the uploaded file + # Example return: + # Dict("status" => 200, "uploadid" => "abc123", "fileid" => "file456", "url" => "http://my-server.com/abc123/file456/myfile.bin") + throw(NotImplementedError("implement upload logic here")) end ``` """ @@ -723,75 +723,6 @@ function _serialize_data(data::Any, payload_type::String) 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 This function processes incoming messages, handling both direct transport (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)`) # 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 - `base_delay::Int = 100` - Initial 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) 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"), ...] + +# 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(