Handle empty message (#54)

* Stop using obsolete Base.write_sub

Fixes #49

* Catching messages without body

Co-authored-by: Sheldon Hearn <[email protected]>

Author: mfalt

src/protocol.jl

@@ -1374,11 +1374,33 @@ function on_basic_get_empty_or_ok(chan::MessageChannel, m::TAMQPMethodFrame, ctx
     nothing
 end
 
+function on_channel_message_completed(chan::MessageChannel, msg::Message)
+    # got all data for msg
+    if isempty(msg.consumer_tag)
+        put!(chan.chan_get, pop!(chan.partial_msgs))
+    else
+        lock(chan.lck) do
+            if msg.consumer_tag in keys(chan.consumers)
+                put!(chan.consumers[msg.consumer_tag].recvq, pop!(chan.partial_msgs))
+            else
+                put!(get!(()->Channel{Message}(typemax(Int)), chan.pending_msgs, msg.consumer_tag), msg)
+                @debug("holding message, no consumer yet with tag", tag=msg.consumer_tag)
+            end
+        end
+    end
+    nothing
+end
+
 function on_channel_message_in(chan::MessageChannel, m::TAMQPContentHeaderFrame, ctx)
     msg = last(chan.partial_msgs)
     msg.properties = m.hdrpayload.proplist
     msg.data = Vector{UInt8}(undef, m.hdrpayload.bodysize)
     msg.filled = 0
+
+    if m.hdrpayload.bodysize == 0
+        # got all data for msg
+        on_channel_message_completed(chan, msg)
+    end
     nothing
 end
 
@@ -1392,18 +1414,7 @@ function on_channel_message_in(chan::MessageChannel, m::TAMQPContentBodyFrame, c
 
     if msg.filled >= length(msg.data)
         # got all data for msg
-        if isempty(msg.consumer_tag)
-            put!(chan.chan_get, pop!(chan.partial_msgs))
-        else
-            lock(chan.lck) do
-                if msg.consumer_tag in keys(chan.consumers)
-                    put!(chan.consumers[msg.consumer_tag].recvq, pop!(chan.partial_msgs))
-                else
-                    put!(get!(()->Channel{Message}(typemax(Int)), chan.pending_msgs, msg.consumer_tag), msg)
-                    @debug("holding message, no consumer yet with tag", tag=msg.consumer_tag)
-                end
-            end
-        end
+        on_channel_message_completed(chan, msg)
     end
 
     nothing

test/test_coverage.jl

@@ -116,9 +116,13 @@ function runtests(;virtualhost="/", host="localhost", port=AMQPClient.AMQP_DEFAU
                 @test rcvd_msg.exchange == EXCG_DIRECT
                 @test rcvd_msg.redelivered == false
                 @test rcvd_msg.routing_key == ROUTE1
-                @test rcvd_msg.data == M.data
                 global msg_count
                 msg_count += 1
+                if msg_count <= 10
+                    @test rcvd_msg.data == M.data
+                else
+                    @test rcvd_msg.data == UInt8[]
+                end
                 println("received msg $(msg_count): $(String(rcvd_msg.data))")
                 basic_ack(chan1, rcvd_msg.delivery_tag)
             end
@@ -137,6 +141,23 @@ function runtests(;virtualhost="/", host="localhost", port=AMQPClient.AMQP_DEFAU
             end
             @test msg_count == 10
 
+            @info("testing empty messages")
+            # Test sending and receiving empty message
+            M_empty = Message(Vector{UInt8}(), content_type="text/plain", delivery_mode=PERSISTENT)
+            basic_publish(chan1, M_empty; exchange=EXCG_DIRECT, routing_key=ROUTE1)
+
+            M_no_ct = Message(Vector{UInt8}(), delivery_mode=PERSISTENT)
+            basic_publish(chan1, M_no_ct; exchange=EXCG_DIRECT, routing_key=ROUTE1)
+
+            println("Waiting")
+            # wait for a reasonable time to receive last two messages
+            for idx in 1:5
+                (msg_count == 12) && break
+                sleep(1)
+            end
+            println("Waited")
+            @test msg_count == 12
+
             # cancel the consumer task
             @test basic_cancel(chan1, consumer_tag)