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-18 20:26:25 +07:00
parent 82804c6803
commit 9c4a9ea1e5
9 changed files with 38 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
new_testFile_large.zip
View File

Binary file not shown.

BIN
new_testFile_small.zip
View File

Binary file not shown.

78
src/NATSBridge.jl
View File

@@ -57,14 +57,14 @@ struct msgPayload_v1
end end
# constructor # constructor
function msgPayload_v1(; function msgPayload_v1(
data::Any,
type::String;
id::String = "", id::String = "",
dataname::String = "", dataname::String = string(uuid4()),
type::String = "text",
transport::String = "direct", transport::String = "direct",
encoding::String = "none", encoding::String = "none",
size::Integer = 0, size::Integer = 0,
data::Any = nothing,
metadata::Dict{String, T} = Dict{String, Any}() metadata::Dict{String, T} = Dict{String, Any}()
) where {T<:Any} ) where {T<:Any}
return msgPayload_v1( return msgPayload_v1(
@@ -101,11 +101,12 @@ struct msgEnvelope_v1
end end
# constructor # constructor
function msgEnvelope_v1(; function msgEnvelope_v1(
sendTo::String,
payloads::AbstractArray{msgPayload_v1};
correlationId::String = "", correlationId::String = "",
msgId::String = "", msgId::String = "",
timestamp::String = "", timestamp::String = string(Dates.now()),
sendTo::String = "",
msgPurpose::String = "", msgPurpose::String = "",
senderName::String = "", senderName::String = "",
senderId::String = "", senderId::String = "",
@@ -114,8 +115,7 @@ function msgEnvelope_v1(;
replyTo::String = "", replyTo::String = "",
replyToMsgId::String = "", replyToMsgId::String = "",
brokerURL::String = DEFAULT_NATS_URL, brokerURL::String = DEFAULT_NATS_URL,
metadata::Dict{String, Any} = Dict{String, Any}(), metadata::Dict{String, Any} = Dict{String, Any}()
payloads::AbstractArray{msgPayload_v1} = msgPayload_v1[]
) )
return msgEnvelope_v1( return msgEnvelope_v1(
correlationId, correlationId,
@@ -309,7 +309,6 @@ function smartsend(
# Generate message metadata # Generate message metadata
msg_id = string(uuid4()) msg_id = string(uuid4())
timestamp = string(Dates.now())
# Process each payload in the list # Process each payload in the list
payloads = msgPayload_v1[] payloads = msgPayload_v1[]
@@ -328,14 +327,14 @@ function smartsend(
# Create msgPayload_v1 for direct transport # Create msgPayload_v1 for direct transport
payload = msgPayload_v1( payload = msgPayload_v1(
id = string(uuid4()), payload_b64,
dataname = dataname, payload_type;
type = payload_type, id = string(uuid4()),
transport = "direct", dataname = dataname,
encoding = "base64", transport = "direct",
size = payload_size, encoding = "base64",
data = payload_b64, size = payload_size,
metadata = Dict{String, Any}("payload_bytes" => payload_size) metadata = Dict{String, Any}("payload_bytes" => payload_size)
) )
push!(payloads, payload) push!(payloads, payload)
else else
@@ -354,14 +353,14 @@ function smartsend(
# Create msgPayload_v1 for link transport # Create msgPayload_v1 for link transport
payload = msgPayload_v1( payload = msgPayload_v1(
id = string(uuid4()), url,
dataname = dataname, payload_type;
type = payload_type, id = string(uuid4()),
transport = "link", dataname = dataname,
encoding = "none", transport = "link",
size = payload_size, encoding = "none",
data = url, size = payload_size,
metadata = Dict{String, Any}() metadata = Dict{String, Any}()
) )
push!(payloads, payload) push!(payloads, payload)
end end
@@ -369,20 +368,19 @@ function smartsend(
# Create msgEnvelope_v1 with all payloads # Create msgEnvelope_v1 with all payloads
env = msgEnvelope_v1( env = msgEnvelope_v1(
correlationId = cid, subject,
msgId = msg_id, payloads;
timestamp = timestamp, correlationId = cid,
sendTo = subject, msgId = msg_id,
msgPurpose = msg_purpose, msgPurpose = msg_purpose,
senderName = sender_name, senderName = sender_name,
senderId = string(uuid4()), senderId = string(uuid4()),
receiverName = receiver_name, receiverName = receiver_name,
receiverId = receiver_id, receiverId = receiver_id,
replyTo = reply_to, replyTo = reply_to,
replyToMsgId = reply_to_msg_id, replyToMsgId = reply_to_msg_id,
brokerURL = nats_url, brokerURL = nats_url,
metadata = Dict{String, Any}(), metadata = Dict{String, Any}(),
payloads = payloads
) )
msg_json = envelope_to_json(env) # Convert envelope to JSON msg_json = envelope_to_json(env) # Convert envelope to JSON

0
test/test_julia_julia_data_transfer_receive.jl → test/julia_to_julia_file_transfer_receiver.jl
View File

0
test/test_julia_julia_data_transfer_send.jl → test/julia_to_julia_file_transfer_sender.jl
View File

67
test/scenario1_command_control.jl
View File

@@ -1,67 +0,0 @@
#!/usr/bin/env julia
# Scenario 1: Command & Control (Small JSON)
# Tests small JSON payloads (< 1MB) sent directly via NATS
using NATS
using JSON3
using UUIDs
# Include the bridge module
include("../src/julia_bridge.jl")
using .BiDirectionalBridge
# Configuration
const CONTROL_SUBJECT = "control"
const RESPONSE_SUBJECT = "control_response"
const NATS_URL = "nats://localhost:4222"
# Create correlation ID for tracing
correlation_id = string(uuid4())
# Receiver: Listen for control commands
function start_control_listener()
conn = NATS.Connection(NATS_URL)
try
NATS.subscribe(conn, CONTROL_SUBJECT) do msg
log_trace(msg.data)
# Parse the envelope
env = MessageEnvelope(String(msg.data))
# Parse JSON payload
config = JSON3.read(env.payload)
# Execute simulation with parameters
step_size = config.step_size
iterations = config.iterations
# Simulate processing
sleep(0.1) # Simulate some work
# Send acknowledgment
response = Dict(
"status" => "Running",
"correlation_id" => env.correlation_id,
"step_size" => step_size,
"iterations" => iterations
)
NATS.publish(conn, RESPONSE_SUBJECT, JSON3.stringify(response))
log_trace("Sent response: $(JSON3.stringify(response))")
end
# Keep listening for 5 seconds
sleep(5)
finally
NATS.close(conn)
end
end
# Helper: Log with correlation ID
function log_trace(message)
timestamp = Dates.now()
println("[$timestamp] [Correlation: $correlation_id] $message")
end
# Run the listener
start_control_listener()

34
test/scenario1_command_control.js
View File

@@ -1,34 +0,0 @@
#!/usr/bin/env node
// Scenario 1: Command & Control (Small JSON)
// Tests small JSON payloads (< 1MB) sent directly via NATS
const { SmartSend } = require('../js_bridge');
// Configuration
const CONTROL_SUBJECT = "control";
const NATS_URL = "nats://localhost:4222";
// Create correlation ID for tracing
const correlationId = require('uuid').v4();
// Sender: Send control command to Julia
async function sendControlCommand() {
const config = {
step_size: 0.01,
iterations: 1000
};
// Send via SmartSend with type="json"
const env = await SmartSend(
CONTROL_SUBJECT,
config,
"json",
{ correlationId }
);
console.log(`Sent control command with correlation_id: ${correlationId}`);
console.log(`Envelope: ${JSON.stringify(env, null, 2)}`);
}
// Run the sender
sendControlCommand().catch(console.error);

66
test/scenario2_large_table.jl
View File

@@ -1,66 +0,0 @@
#!/usr/bin/env julia
# Scenario 2: Deep Dive Analysis (Large Arrow Table)
# Tests large Arrow tables (> 1MB) sent via HTTP fileserver
using NATS
using Arrow
using DataFrames
using JSON3
using UUIDs
# Include the bridge module
include("../src/julia_bridge.jl")
using .BiDirectionalBridge
# Configuration
const ANALYSIS_SUBJECT = "analysis_results"
const RESPONSE_SUBJECT = "analysis_response"
const NATS_URL = "nats://localhost:4222"
# Create correlation ID for tracing
correlation_id = string(uuid4())
# Receiver: Listen for analysis results
function start_analysis_listener()
conn = NATS.Connection(NATS_URL)
try
NATS.subscribe(conn, ANALYSIS_SUBJECT) do msg
log_trace("Received message from $(msg.subject)")
# Parse the envelope
env = MessageEnvelope(String(msg.data))
# Use SmartReceive to handle the data
result = SmartReceive(msg)
# Process the data based on type
if result.envelope.type == "table"
df = result.data
log_trace("Received DataFrame with $(nrows(df)) rows")
log_trace("DataFrame columns: $(names(df))")
# Send acknowledgment
response = Dict(
"status" => "Processed",
"correlation_id" => env.correlation_id,
"row_count" => nrows(df)
)
NATS.publish(conn, RESPONSE_SUBJECT, JSON3.stringify(response))
end
end
# Keep listening for 10 seconds
sleep(10)
finally
NATS.close(conn)
end
end
# Helper: Log with correlation ID
function log_trace(message)
timestamp = Dates.now()
println("[$timestamp] [Correlation: $correlation_id] $message")
end
# Run the listener
start_analysis_listener()

54
test/scenario2_large_table.js
View File

@@ -1,54 +0,0 @@
#!/usr/bin/env node
// Scenario 2: Deep Dive Analysis (Large Arrow Table)
// Tests large Arrow tables (> 1MB) sent via HTTP fileserver
const { SmartSend } = require('../js_bridge');
// Configuration
const ANALYSIS_SUBJECT = "analysis_results";
const NATS_URL = "nats://localhost:4222";
// Create correlation ID for tracing
const correlationId = require('uuid').v4();
// Sender: Send large Arrow table to Julia
async function sendLargeTable() {
// Create a large DataFrame-like structure (10 million rows)
// For testing, we'll create a smaller but still large table
const numRows = 1000000; // 1 million rows
const data = {
id: Array.from({ length: numRows }, (_, i) => i + 1),
value: Array.from({ length: numRows }, () => Math.random()),
category: Array.from({ length: numRows }, () => ['A', 'B', 'C'][Math.floor(Math.random() * 3)])
};
// Convert to Arrow Table
const { Table, Vector, RecordBatch } = require('apache-arrow');
const idVector = Vector.from(data.id);
const valueVector = Vector.from(data.value);
const categoryVector = Vector.from(data.category);
const table = Table.from({
id: idVector,
value: valueVector,
category: categoryVector
});
// Send via SmartSend with type="table"
const env = await SmartSend(
ANALYSIS_SUBJECT,
table,
"table",
{ correlationId }
);
console.log(`Sent large table with ${numRows} rows`);
console.log(`Correlation ID: ${correlationId}`);
console.log(`Transport: ${env.transport}`);
console.log(`URL: ${env.url || 'N/A'}`);
}
// Run the sender
sendLargeTable().catch(console.error);