|
|
|
|
@@ -436,7 +436,6 @@ function smartsend(
|
|
|
|
|
|
|
|
|
|
# Generate correlation ID if not provided
|
|
|
|
|
cid = correlation_id !== nothing ? correlation_id : string(uuid4()) # Create or use provided correlation ID
|
|
|
|
|
|
|
|
|
|
log_trace(cid, "Starting smartsend for subject: $subject") # Log start of send operation
|
|
|
|
|
|
|
|
|
|
# Generate message metadata
|
|
|
|
|
@@ -706,7 +705,7 @@ A HTTP file server is required along with its download function.
|
|
|
|
|
- `max_delay::Int = 5000` - Maximum delay for exponential backoff in ms
|
|
|
|
|
|
|
|
|
|
# Return:
|
|
|
|
|
- `Vector{Tuple{String, Any, String}}` - List of (dataname, data, type) tuples
|
|
|
|
|
- JSON object of envelope with list of (dataname, data, data_type) tuples in payloads field
|
|
|
|
|
|
|
|
|
|
# Example
|
|
|
|
|
```jldoctest
|
|
|
|
|
@@ -724,22 +723,22 @@ function smartreceive(
|
|
|
|
|
max_delay::Int = 5000
|
|
|
|
|
)
|
|
|
|
|
# Parse the JSON envelope
|
|
|
|
|
json_data = JSON.parse(String(msg.payload))
|
|
|
|
|
log_trace(json_data["correlation_id"], "Processing received message") # Log message processing start
|
|
|
|
|
env_json_obj = JSON.parse(String(msg.payload))
|
|
|
|
|
log_trace(env_json_obj["correlation_id"], "Processing received message") # Log message processing start
|
|
|
|
|
|
|
|
|
|
# Process all payloads in the envelope
|
|
|
|
|
payloads_list = Tuple{String, Any, String}[]
|
|
|
|
|
|
|
|
|
|
# Get number of payloads
|
|
|
|
|
num_payloads = length(json_data["payloads"])
|
|
|
|
|
num_payloads = length(env_json_obj["payloads"])
|
|
|
|
|
|
|
|
|
|
for i in 1:num_payloads
|
|
|
|
|
payload = json_data["payloads"][i]
|
|
|
|
|
payload = env_json_obj["payloads"][i]
|
|
|
|
|
transport = String(payload["transport"])
|
|
|
|
|
dataname = String(payload["dataname"])
|
|
|
|
|
|
|
|
|
|
if transport == "direct" # Direct transport - payload is in the message
|
|
|
|
|
log_trace(json_data["correlation_id"], "Direct transport - decoding payload '$dataname'") # Log direct transport handling
|
|
|
|
|
log_trace(env_json_obj["correlation_id"], "Direct transport - decoding payload '$dataname'") # Log direct transport handling
|
|
|
|
|
|
|
|
|
|
# Extract base64 payload from the payload
|
|
|
|
|
payload_b64 = String(payload["data"])
|
|
|
|
|
@@ -749,28 +748,28 @@ function smartreceive(
|
|
|
|
|
|
|
|
|
|
# Deserialize based on type
|
|
|
|
|
data_type = String(payload["payload_type"])
|
|
|
|
|
data = _deserialize_data(payload_bytes, data_type, json_data["correlation_id"])
|
|
|
|
|
data = _deserialize_data(payload_bytes, data_type, env_json_obj["correlation_id"])
|
|
|
|
|
|
|
|
|
|
push!(payloads_list, (dataname, data, data_type))
|
|
|
|
|
elseif transport == "link" # Link transport - payload is at URL
|
|
|
|
|
# Extract download URL from the payload
|
|
|
|
|
url = String(payload["data"])
|
|
|
|
|
log_trace(json_data["correlation_id"], "Link transport - fetching '$dataname' from URL: $url") # Log link transport handling
|
|
|
|
|
log_trace(env_json_obj["correlation_id"], "Link transport - fetching '$dataname' from URL: $url") # Log link transport handling
|
|
|
|
|
|
|
|
|
|
# Fetch with exponential backoff using the download handler
|
|
|
|
|
downloaded_data = fileserver_download_handler(url, max_retries, base_delay, max_delay, json_data["correlation_id"])
|
|
|
|
|
downloaded_data = fileserver_download_handler(url, max_retries, base_delay, max_delay, env_json_obj["correlation_id"])
|
|
|
|
|
|
|
|
|
|
# Deserialize based on type
|
|
|
|
|
data_type = String(payload["payload_type"])
|
|
|
|
|
data = _deserialize_data(downloaded_data, data_type, json_data["correlation_id"])
|
|
|
|
|
data = _deserialize_data(downloaded_data, data_type, env_json_obj["correlation_id"])
|
|
|
|
|
|
|
|
|
|
push!(payloads_list, (dataname, data, data_type))
|
|
|
|
|
else # Unknown transport type
|
|
|
|
|
error("Unknown transport type for payload '$dataname': $(transport)") # Throw error for unknown transport
|
|
|
|
|
end
|
|
|
|
|
end
|
|
|
|
|
json_data["payloads"] = payloads_list
|
|
|
|
|
return json_data # Return envelope with list of (dataname, data, data_type) tuples in payloads field
|
|
|
|
|
env_json_obj["payloads"] = payloads_list
|
|
|
|
|
return env_json_obj # JSON object of envelope with list of (dataname, data, data_type) tuples in payloads field
|
|
|
|
|
end
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@@ -929,7 +928,7 @@ retrieves an upload ID and token, then uploads the file data as multipart form d
|
|
|
|
|
```jldoctest
|
|
|
|
|
using HTTP, JSON
|
|
|
|
|
|
|
|
|
|
file_server_url = "http://localhost:8080"
|
|
|
|
|
fileserver_url = "http://localhost:8080"
|
|
|
|
|
filename = "test.txt"
|
|
|
|
|
data = Vector{UInt8}("hello world")
|
|
|
|
|
|
|
|
|
|
@@ -1006,7 +1005,7 @@ retrieves an upload ID and token, then uploads the file data as multipart form d
|
|
|
|
|
```jldoctest
|
|
|
|
|
using HTTP, JSON
|
|
|
|
|
|
|
|
|
|
file_server_url = "http://localhost:8080"
|
|
|
|
|
fileserver_url = "http://localhost:8080"
|
|
|
|
|
filepath = "./test.zip"
|
|
|
|
|
|
|
|
|
|
# Upload to local plik server
|
|
|
|
|
|