Merge pull request 'test_smartreceive_function' (#1) from test_smartreceive_function into main

Reviewed-on: #1

src/NATSBridge.jl

@@ -104,7 +104,7 @@ function log_trace(correlation_id::String, message::String)
     timestamp = Dates.now()  # Get current timestamp
     @info "[$timestamp] [Correlation: $correlation_id] $message"  # Log formatted message
 end
-2112
+
 
 """ smartsend - Send data either directly via NATS or via a fileserver URL, depending on payload size
 
@@ -162,7 +162,7 @@ function smartsend(
   subject::String,  # smartreceive's subject
   data::Any,
   type::String = "json";
-  dataname=string(UUIDs.uuid4()),
+  dataname="NA",
   nats_url::String = DEFAULT_NATS_URL,
   fileserver_url::String = DEFAULT_FILESERVER_URL,
   fileServerUploadHandler::Function=plik_oneshot_upload,  # a function to handle uploading data to specific HTTP fileserver
@@ -191,7 +191,7 @@ function smartsend(
           type = type,
           transport = "direct",
           payload = payload_b64,
-          metadata = Dict("content_length" => payload_size, "format" => "arrow_ipc_stream")
+          metadata = Dict("dataname" => dataname, "content_length" => payload_size, "format" => "arrow_ipc_stream")
       )
       
       msg_json = envelope_to_json(env)  # Convert envelope to JSON
@@ -217,7 +217,7 @@ function smartsend(
           type = type,
           transport = "link",
           url = url,
-          metadata = Dict("content_length" => payload_size, "format" => "arrow_ipc_stream")
+          metadata = Dict("dataname" => dataname, "content_length" => payload_size, "format" => "arrow_ipc_stream")
       )
       
       msg_json = envelope_to_json(env)  # Convert envelope to JSON
@@ -364,12 +364,12 @@ function smartreceive(
         log_trace(env.correlation_id, "Link transport - fetching from URL")  # Log link transport handling
         
         # Fetch with exponential backoff
-        data = _fetch_with_backoff(env.url, max_retries, base_delay, max_delay, env.correlation_id)  # Fetch data from URL
+        downloaded_data = _fetch_with_backoff(env.url, max_retries, base_delay, max_delay, env.correlation_id)  # Fetch data from URL
         
         # Deserialize based on type
-        result = _deserialize_data(data, env.type, env.correlation_id, env.metadata)  # Convert bytes to Julia data
+        data = _deserialize_data(downloaded_data, env.type, env.correlation_id, env.metadata)  # Convert bytes to Julia data
         
-        return (data = result, envelope = env)  # Return data and envelope as tuple
+        return (data = data, envelope = env)  # Return data and envelope as tuple
     else  # Unknown transport type
         error("Unknown transport type: $(env.transport)")  # Throw error for unknown transport
     end

test/scenario3_julia_to_julia.jl

@@ -1,66 +0,0 @@
-#!/usr/bin/env julia
-# Scenario 3: Julia-to-Julia Service Communication
-# Tests bi-directional communication between two Julia services
-
-using NATS
-using Arrow
-using DataFrames
-using JSON3
-using UUIDs
-
-# Include the bridge module
-include("../src/julia_bridge.jl")
-using .BiDirectionalBridge
-
-# Configuration
-const SUBJECT1 = "julia_to_js"
-const SUBJECT2 = "js_to_julia"
-const RESPONSE_SUBJECT = "response"
-const NATS_URL = "nats://localhost:4222"
-
-# Create correlation ID for tracing
-correlation_id = string(uuid4())
-
-# Julia-to-Julia Test: Large Arrow Table
-function test_julia_to_julia_large_table()
-    conn = NATS.Connection(NATS_URL)
-    try
-        # Subscriber on SUBJECT2 to receive data from Julia sender
-        NATS.subscribe(conn, SUBJECT2) do msg
-            log_trace("[$(Dates.now())] Received on $SUBJECT2")
-            
-            # Use SmartReceive to handle the data
-            result = SmartReceive(msg)
-            
-            # Check transport type
-            if result.envelope.transport == "direct"
-                log_trace("Received direct transport with $(length(result.data)) bytes")
-            else
-                # For link transport, result.data is the URL
-                log_trace("Received link transport at $(result.data)")
-            end
-            
-            # Send response back
-            response = Dict(
-                "status" => "Processed",
-                "correlation_id" => result.envelope.correlation_id,
-                "timestamp" => Dates.now()
-            )
-            NATS.publish(conn, RESPONSE_SUBJECT, JSON3.stringify(response))
-        end
-        
-        # Keep listening
-        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 test
-test_julia_to_julia_large_table()

test/scenario_tests.md

@@ -1,148 +0,0 @@
-# Test Scenarios for Bi-Directional Data Bridge
-
-## Scenario 1: Command & Control (Small JSON)
-Tests small JSON payloads (< 1MB) sent directly via NATS.
-
-### Julia (Receiver)
-```julia
-using NATS
-using JSON3
-
-# Subscribe to control subject
-subscribe(nats, "control") do msg
-    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
-    
-    # Send acknowledgment
-    response = Dict("status" => "Running", "correlation_id" => env.correlation_id)
-    publish(nats, "control_response", JSON3.stringify(response))
-end
-```
-
-### JavaScript (Sender)
-```javascript
-const { SmartSend } = require('./js_bridge');
-
-// Create small JSON config
-const config = {
-    step_size: 0.01,
-    iterations: 1000
-};
-
-// Send via SmartSend with type="json"
-await SmartSend("control", config, "json");
-```
-
-## Scenario 2: Deep Dive Analysis (Large Arrow Table)
-Tests large Arrow tables (> 1MB) sent via HTTP fileserver.
-
-### Julia (Sender)
-```julia
-using Arrow
-using DataFrames
-
-# Create large DataFrame (500MB, 10 million rows)
-df = DataFrame(
-    id = 1:10_000_000,
-    value = rand(10_000_000),
-    category = rand(["A", "B", "C"], 10_000_000)
-)
-
-# Convert to Arrow IPC stream and send
-await SmartSend("analysis_results", df, "table");
-```
-
-### JavaScript (Receiver)
-```javascript
-const { SmartReceive } = require('./js_bridge');
-
-// Receive message with URL
-const result = await SmartReceive(msg);
-
-// Fetch data from HTTP server
-const table = result.data;
-
-// Load into Perspective.js or D3
-// Use table data for visualization
-```
-
-## Scenario 3: Live Binary Processing
-Tests binary data (binary) sent from JS to Julia for FFT/transcription.
-
-### JavaScript (Sender)
-```javascript
-const { SmartSend } = require('./js_bridge');
-
-// Capture binary chunk (2 seconds, 44.1kHz, 1 channel)
-const binaryData = await navigator.mediaDevices.getUserMedia({ binary: true });
-
-// Send as binary with metadata headers
-await SmartSend("binary_input", binaryData, "binary", {
-    metadata: {
-        sample_rate: 44100,
-        channels: 1
-    }
-});
-```
-
-### Julia (Receiver)
-```julia
-using WAV
-using DSP
-
-# Receive binary data
-function process_binary(data)
-    # Perform FFT or AI transcription
-    spectrum = fft(data)
-    
-    # Send results back (JSON + Arrow table)
-    results = Dict("transcription" => "sample text", "spectrum" => spectrum)
-    await SmartSend("binary_output", results, "json")
-end
-```
-
-## Scenario 4: Catch-Up (JetStream)
-Tests temporal decoupling with NATS JetStream.
-
-### Julia (Producer)
-```julia
-# Publish to JetStream
-using NATS
-
-function publish_health_status(nats)
-    jetstream = JetStream(nats, "health_updates")
-    
-    while true
-        status = Dict("cpu" => rand(), "memory" => rand())
-        publish(jetstream, "health", status)
-        sleep(5)  # Every 5 seconds
-    end
-end
-```
-
-### JavaScript (Consumer)
-```javascript
-const { connect } = require('nats');
-
-const nc = await connect({ servers: ['nats://localhost:4222'] });
-const js = nc.jetstream();
-
-// Request replay from last 10 minutes
-const consumer = await js.pullSubscribe("health", { 
-    durable_name: "catchup",
-    max_batch: 100,
-    max_ack_wait: 30000
-});
-
-// Process historical and real-time messages
-for await (const msg of consumer) {
-    const result = await SmartReceive(msg);
-    // Process the data
-    msg.ack();
-}

test/test_julia_julia_data_transfer.jl

@@ -16,9 +16,14 @@ const FILESERVER_URL = "http://192.168.88.104:8080"
 # Create correlation ID for tracing
 correlation_id = string(uuid4())
 
+
+# ------------------------------------------------------------------------------------------------ #
+#                                        test file transfer                                        #
+# ------------------------------------------------------------------------------------------------ #
+
 # File path for large binary payload test
-const LARGE_FILE_PATH = "./testFile_small.zip"
+const FILE_PATH = "./testFile_small.zip"
-const filename = basename(LARGE_FILE_PATH)
+const filename = basename(FILE_PATH)
 
 # Helper: Log with correlation ID
 function log_trace(message)
@@ -30,8 +35,8 @@ end
 function test_large_binary_send()
     conn = NATS.connect(NATS_URL)
     # Read the large file as binary data
-    log_trace("Reading large file: $LARGE_FILE_PATH")
+    log_trace("Reading large file: $FILE_PATH")
-    file_data = read(LARGE_FILE_PATH)
+    file_data = read(FILE_PATH)
     
     file_size = length(file_data)
     log_trace("File size: $file_size bytes")
@@ -69,13 +74,12 @@ function test_large_binary_receive()
     
     # Use NATSBridge.smartreceive to handle the data
     result = NATSBridge.smartreceive(msg)
-    # println("envelope----- ", result.envelope)
     # Check transport type
     if result.envelope.transport == "direct"
-        log_trace("Received direct transport with ---- bytes")
+        log_trace("Received direct transport")
     else
         # For link transport, result.data is the URL
-        log_trace("Received link transport at ---")
+        log_trace("Received link transport")
     end
               
     # Verify the received data matches the original
@@ -85,40 +89,43 @@ function test_large_binary_receive()
         log_trace("Received $(file_size) bytes of binary data")
         
         # Save received data to a test file
-        #[WORKING] add dataname so I know it is a file
+        println("metadata ", result.envelope.metadata)
-        filename = basename(result.envelope.url)
+        dataname = result.envelope.metadata["dataname"]
-        output_path = "./new_$filename"
+        if dataname != "NA"
+          output_path = "./new_$dataname"
           write(output_path, result.data)
           log_trace("Saved received data to $output_path")
+        end
         
         # Verify file size
-        original_size = length(read(LARGE_FILE_PATH))
+        original_size = length(read(FILE_PATH))
-        if file_size == original_size
+        if file_size == result.envelope.metadata["content_length"]
-            log_trace("SUCCESS: File size matches! Original: $original_size bytes")
+            log_trace("SUCCESS: File size matches! Original: $(result.envelope.metadata["content_length"]) bytes")
         else
-            log_trace("WARNING: File size mismatch! Original: $original_size, Received: $file_size")
+            log_trace("WARNING: File size mismatch! Original: $(result.envelope.metadata["content_length"]), Received: $file_size")
         end
       end
     end
   end
 
   # Keep listening for 10 seconds
-  sleep(60)  
+  sleep(120)  
   NATS.drain(conn)
 end
 
+
 # Run the test
 println("Starting large binary payload test...")
 println("Correlation ID: $correlation_id")
-println("Large file: $LARGE_FILE_PATH")
+println("File: $FILE_PATH")
 
-# # Run sender first
+# Run sender first
-# println("start smartsend")
+println("start smartsend")
-# test_large_binary_send()
+test_large_binary_send()
 
-# Run receiver
+# # Run receiver
-println("testing smartreceive")
+# println("testing smartreceive")
-test_large_binary_receive()
+# test_large_binary_receive()
 
 println("Test completed.")
 
@@ -137,3 +144,47 @@ println("Test completed.")
 
 
 
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+