check stale conn while chunking message to send

This fixes a condition where sending a message on a stale connection can potentially get into an infinite loop while chunking a large message.

Author: tanmaykm

Project.toml

@@ -3,7 +3,7 @@ uuid = "79c8b4cd-a41a-55fa-907c-fab5288e1383"
 keywords = ["amqpclient", "rabbitmq", "amqp", "amqp-client", "message-queue"]
 license = "MIT"
 desc = "A Julia AMQP (Advanced Message Queuing Protocol) / RabbitMQ Client."
-version = "0.4.2"
+version = "0.4.3"
 
 [deps]
 Logging = "56ddb016-857b-54e1-b83d-db4d58db5568"

src/protocol.jl

@@ -267,13 +267,15 @@ 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)
     try
         put!(c.sendq, TAMQPGenericFrame(f))
         for m in msgframes
             put!(c.sendq, TAMQPGenericFrame(m))
         end
     finally
+        @debug("queued messageframes", nqueued=length(c.sendq.data))
         put!(c.sendlck, lck)
     end
     nothing
@@ -292,17 +294,25 @@ function send(c::MessageChannel, payload::TAMQPMethodPayload, msg::Union{Message
         # 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
+
         while offset <= msglen
-            msgend = min(msglen, offset + c.conn.framemax - 1)
+            msgend = min(msglen, offset + framemax - 1)
             bodypayload = TAMQPBodyPayload(message.data[offset:msgend])
             offset = msgend + 1
+            @debug("sending", msglen, offset)
             push!(msgframes, TAMQPContentBodyFrame(frameprop, bodypayload))
         end
 
         send(c, TAMQPMethodFrame(frameprop, payload), msgframes)
     else
         send(c, TAMQPMethodFrame(frameprop, payload))
     end
+    @debug("sent", methodname=method_name(payload))
 end
 
 # ----------------------------------------