reduce copying of large messages while sending

src/protocol.jl

@@ -287,55 +287,49 @@ isopen(c::MessageChannel) = isopen(c.conn) && (c.id in keys(c.conn.channels))
 get_property(c::MessageChannel, s::Symbol, default) = get_property(c.conn, s, default)
 get_property(c::Connection, s::Symbol, default) = get(c.properties, s, default)
 
-send(c::MessageChannel, f, msgframes::Vector=[]) = send(c.conn, f, msgframes)
-function send(c::Connection, f, msgframes::Vector=[])
-    #uncomment to enable synchronization (not required till we have preemptive tasks or threads)
-    @debug("queing messageframes", nframes=length(msgframes))
-    lck = take!(c.sendlck)
+with_sendlock(f, c::MessageChannel) = with_sendlock(f, c.conn)
+with_sendlock(f, c::Connection) = with_sendlock(f, c.sendlck)
+function with_sendlock(f, sendlck::Channel{UInt8})
+    lck = take!(sendlck)
     try
-        put!(c.sendq, TAMQPGenericFrame(f))
-        for m in msgframes
-            put!(c.sendq, TAMQPGenericFrame(m))
-        end
+        f()
     finally
-        @debug("queued messageframes", nqueued=length(c.sendq.data))
-        put!(c.sendlck, lck)
+        put!(sendlck, lck)
     end
-    nothing
 end
-function send(c::MessageChannel, payload::TAMQPMethodPayload, msg::Union{Message, Nothing}=nothing)
-    @debug("sending", methodname=method_name(payload), hascontent=(msg !== nothing))
+send(c::MessageChannel, f) = send(c.conn, f)
+send(c::Connection, f) = put!(c.sendq, TAMQPGenericFrame(f))
+function send(c::MessageChannel, payload::TAMQPMethodPayload)
+    @debug("sending without content", methodname=method_name(payload))
+    frameprop = TAMQPFrameProperties(c.id,0)
+    send(c, TAMQPMethodFrame(frameprop, payload))
+end
+function send(c::MessageChannel, payload::TAMQPMethodPayload, msg::Message)
+    @debug("sending with content", methodname=method_name(payload))
     frameprop = TAMQPFrameProperties(c.id,0)
-    if msg !== nothing
-        msgframes = []
-        message = msg
+    framemax = c.conn.framemax
+    if framemax <= 0
+        errormsg = (c.conn.state == CONN_STATE_OPEN) ? "Unexpected framemax ($framemax) value for connection" : "Connection closed"
+        throw(AMQPClientException(errormsg))
+    end
 
-        # send message header frame
-        hdrpayload = TAMQPHeaderPayload(payload.class, message)
-        push!(msgframes, TAMQPContentHeaderFrame(frameprop, hdrpayload))
+    with_sendlock(c) do
+        send(c, TAMQPMethodFrame(frameprop, payload))
+        hdrpayload = TAMQPHeaderPayload(payload.class, msg)
+        send(c, TAMQPContentHeaderFrame(frameprop, hdrpayload))
 
         # send one or more message body frames
         offset = 1
-        msglen = length(message.data)
-        framemax = c.conn.framemax
-        if framemax <= 0
-            errormsg = (c.conn.state == CONN_STATE_OPEN) ? "Unexpected framemax ($framemax) value for connection" : "Connection closed"
-            throw(AMQPClientException(errormsg))
-        end
-
+        msglen = length(msg.data)
+        @debug("sending message with content body", msglen)
         while offset <= msglen
             msgend = min(msglen, offset + framemax - 1)
-            bodypayload = TAMQPBodyPayload(message.data[offset:msgend])
+            bodypayload = TAMQPBodyPayload(msg.data[offset:msgend])
             offset = msgend + 1
-            @debug("sending", msglen, offset)
-            push!(msgframes, TAMQPContentBodyFrame(frameprop, bodypayload))
+            @debug("sending content body frame", msglen, offset)
+            send(c, TAMQPContentBodyFrame(frameprop, bodypayload))
         end
-
-        send(c, TAMQPMethodFrame(frameprop, payload), msgframes)
-    else
-        send(c, TAMQPMethodFrame(frameprop, payload))
     end
-    @debug("sent", methodname=method_name(payload))
 end
 
 # ----------------------------------------

src/types.jl

@@ -76,6 +76,7 @@ end
 TAMQPFieldValue(v::T) where {T} = TAMQPFieldValue{T}(FieldIndicatorMap[T], v)
 TAMQPFieldValue(v::Dict) = TAMQPFieldValue(TAMQPFieldTable(v))
 TAMQPFieldValue(v::String) = TAMQPFieldValue(TAMQPLongStr(v))
+TAMQPFieldValue(v::Bool) = TAMQPFieldValue('b', TAMQPBool(v))
 
 struct TAMQPFieldValuePair{T <: TAMQPFV}
     name::TAMQPFieldName

test/runtests.jl

@@ -16,12 +16,12 @@ include("test_rpc.jl")
                 end
             end
         end
-        # @testset "Throughput" begin
-        #     AMQPTestThroughput.runtests()
-        # end
-        # @testset "RPC" begin
-        #     AMQPTestRPC.runtests()
-        # end
+        @testset "Throughput" begin
+            AMQPTestThroughput.runtests()
+        end
+        @testset "RPC" begin
+            AMQPTestRPC.runtests()
+        end
     end
 
     if length(ARGS) > 0

test/test_coverage.jl

@@ -239,6 +239,31 @@ function test_types()
     mframe = AMQPClient.TAMQPMethodFrame(mfprop, mpayload)
     show(iob, mframe)
     @test length(take!(iob)) > 0
+
+    fields = AMQPClient.TAMQPFieldValue[
+        AMQPClient.TAMQPFieldValue(true),
+        AMQPClient.TAMQPFieldValue(1.1),
+        AMQPClient.TAMQPFieldValue(1),
+        AMQPClient.TAMQPFieldValue("hello world"),
+        AMQPClient.TAMQPFieldValue(Dict{String,Int}("one"=>1, "two"=>2)),
+    ]
+
+    fieldarray = AMQPClient.TAMQPFieldArray(fields)
+    simplified_fields = AMQPClient.simplify(fieldarray)
+    @test simplified_fields == Any[
+        0x01,
+        1.1,
+        1,
+         "hello world",
+         Dict{String, Any}("two" => 2, "one" => 1)
+    ]
+
+    iob = PipeBuffer()
+    write(iob, hton(AMQPClient.TAMQPLongUInt(10)))
+    write(iob, UInt8[1,2,3,4,5,6,7,8,9,0])
+    barr = read(iob, AMQPClient.TAMQPByteArray)
+    @test barr.len == 10
+    @test barr.data == UInt8[1,2,3,4,5,6,7,8,9,0]
 end
 
 end # module AMQPTestCoverage