ton/msghandler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix string error

Browse Source
This commit is contained in:
narawat
2026-07-05 19:46:04 +07:00
parent 96e0105511
commit a10e4ddb89
1 changed files with 24 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/msghandler.jl
+23 -14
View File
@@ -446,25 +446,26 @@ function smartpack(
)::Tuple{msg_envelope_v1, String} where {T1<:Any}
# Log start of send operation
log_trace(correlation_id, "Starting smartpack for subject: $subject")
# log_trace(correlation_id, "Starting smartpack for subject: $subject")
# Process each payload in the list
payloads = msg_payload_v1[]
for (dataname, payload_data, payload_type) in data
# @show dataname typeof(payload_data)
@info "msghandler smartpack() 1" @__LINE__
# Serialize data based on type. use bytes as medium for every datatype
payload_bytes = _serialize_data(payload_data, payload_type)
@info "msghandler smartpack() 2" @__LINE__
payload_size = length(payload_bytes) # Calculate payload size in bytes
log_trace(correlation_id, "Serialized payload '$dataname' (payload_type: $payload_type) size: $payload_size bytes") # Log payload size
# log_trace(correlation_id, "Serialized payload '$dataname' (payload_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 include in message envelope
payload_b64 = Base64.base64encode(payload_bytes) # Encode bytes as base64 string
log_trace(correlation_id, "Using direct transport for $payload_size bytes") # Log transport choice
# log_trace(correlation_id, "Using direct transport for $payload_size bytes") # Log transport choice
@info "msghandler smartpack() 3" @__LINE__
# Determine encoding based on payload_type
encoding = "base64"
if payload_type == "jsontable"
@@ -486,8 +487,9 @@ function smartpack(
)
push!(payloads, payload)
else
@info "msghandler smartpack() 4" @__LINE__
# Link path - Upload to HTTP server, include URL in message envelope
log_trace(correlation_id, "Using link transport, uploading to fileserver") # Log link transport choice
# log_trace(correlation_id, "Using link transport, uploading to fileserver") # Log link transport choice
# Upload to HTTP server
response = fileserver_upload_handler(fileserver_url, dataname, payload_bytes)
@@ -497,7 +499,7 @@ function smartpack(
end
url = response["url"] # URL for the uploaded data
log_trace(correlation_id, "Uploaded to URL: $url") # Log successful upload
# log_trace(correlation_id, "Uploaded to URL: $url") # Log successful upload
# Determine encoding based on payload_type
encoding = "none"
@@ -520,6 +522,7 @@ function smartpack(
)
push!(payloads, payload)
end
@info "msghandler smartpack() 5" @__LINE__
end
# Create msg_envelope_v1 with all payloads
@@ -549,7 +552,7 @@ function smartpack(
# # Publish message to NATS using existing connection
# publish_message(NATS_connection, subject, env_json_str, correlation_id)
# end
@info "msghandler smartpack() 6" @__LINE__
return (env, env_json_str)
end
@@ -636,22 +639,22 @@ function _serialize_data(data::Any, payload_type::String)
json_obj = JSON.parse(json_str_2)
"""
if payload_type == "text" # Text data - convert to UTF-8 bytes
if isa(data, String)
data_bytes = Vector{UInt8}(data) # Convert string to UTF-8 bytes
if data <: AbstractString # Text data - convert to UTF-8 bytes
@info "msghandler _serialize_data() 1" @__LINE__
data_bytes = Vector{UInt8}(string(data)) # Convert string to UTF-8 bytes
return data_bytes
else
error("Text data must be a String")
end
elseif payload_type == "dictionary" # JSON data - serialize directly
@info "msghandler _serialize_data() 2" @__LINE__
json_str = JSON.json(data) # Convert Julia data to JSON string
json_str_bytes = Vector{UInt8}(json_str) # Convert JSON string to bytes
return json_str_bytes
elseif payload_type == "arrowtable" # Arrow table data - convert to Arrow IPC stream
@info "msghandler _serialize_data() 3" @__LINE__
io = IOBuffer() # Create in-memory buffer
Arrow.write(io, data) # Write data as Arrow IPC stream to buffer
return take!(io) # Return the buffer contents as bytes
elseif payload_type == "jsontable" # JSON table data - convert to JSON
@info "msghandler _serialize_data() 4" @__LINE__
# data can be Vector{NamedTuple}, Vector{Dict}, or DataFrame
# If DataFrame, convert to Vector{Dict} first
if isa(data, DataFrame)
@@ -667,29 +670,34 @@ function _serialize_data(data::Any, payload_type::String)
json_str = JSON.json(rows)
return Vector{UInt8}(json_str)
else
@info "msghandler _serialize_data() 5" @__LINE__
# Already Vector{NamedTuple} or Vector{Dict}
json_str = JSON.json(data)
return Vector{UInt8}(json_str)
end
elseif payload_type == "image" # Image data - treat as binary
@info "msghandler _serialize_data() 6" @__LINE__
if isa(data, Vector{UInt8})
return data # Return binary data directly
else
error("Image data must be Vector{UInt8}")
end
elseif payload_type == "audio" # Audio data - treat as binary
@info "msghandler _serialize_data() 7" @__LINE__
if isa(data, Vector{UInt8})
return data # Return binary data directly
else
error("Audio data must be Vector{UInt8}")
end
elseif payload_type == "video" # Video data - treat as binary
@info "msghandler _serialize_data() 8" @__LINE__
if isa(data, Vector{UInt8})
return data # Return binary data directly
else
error("Video data must be Vector{UInt8}")
end
elseif payload_type == "binary" # Binary data - treat as binary
@info "msghandler _serialize_data() 9" @__LINE__
if isa(data, IOBuffer) # Check if data is an IOBuffer
return take!(data) # Return buffer contents as bytes
elseif isa(data, Vector{UInt8}) # Check if data is already binary
@@ -698,6 +706,7 @@ function _serialize_data(data::Any, payload_type::String)
error("Binary data must be binary (Vector{UInt8} or IOBuffer)")
end
else # Unknown type
@info "msghandler _serialize_data() 10" @__LINE__
error("Unknown payload_type: $payload_type")
end
end