Merge pull request #29 from JuliaComputing/tan/misc

throw when channel is closed due to server error

Author: tanmaykm

src/protocol.jl

@@ -631,13 +631,6 @@ function _wait_resp(sendmethod, chan::MessageChannel, default_result::T,
     result = default_result
     if !nowait
         reply = Channel{T}(1)
-        # timer to time the request out, in case of an error
-        t = Timer(timeout) do t
-            try
-                put!(reply, timeout_result)
-            catch
-            end
-        end
         # register a callback
         handle(chan, resp_class, resp_meth, resp_handler, reply)
     end
@@ -646,7 +639,18 @@ function _wait_resp(sendmethod, chan::MessageChannel, default_result::T,
 
     if !nowait
         # wait for response
-        result = take!(reply)
+        result = timeout_result
+        if :ok === timedwait(()->(isready(reply) || !isopen(chan)), Float64(timeout); pollint=0.01)
+            if isready(reply)
+                result = take!(reply)
+            else
+                error_message = "Connection closed"
+                if nothing !== chan.closereason
+                    error_message = string(error_message, " - ", string(chan.closereason.code), " (", convert(String, chan.closereason.msg), ")")
+                end
+                throw(AMQPClientException(error_message))
+            end
+        end
         close(reply)
     end
     result

test/test_coverage.jl

@@ -35,6 +35,15 @@ function runtests(;virtualhost="/", host="localhost", port=AMQPClient.AMQP_DEFAU
     testlog("creating exchanges...")
     @test exchange_declare(chan1, EXCG_DIRECT, EXCHANGE_TYPE_DIRECT; arguments=Dict{String,Any}("Hello"=>"World", "Foo"=>"bar"))
     @test exchange_declare(chan1, EXCG_FANOUT, EXCHANGE_TYPE_FANOUT)
+    # redeclaring the exchange with same attributes should be fine
+    @test exchange_declare(chan1, EXCG_FANOUT, EXCHANGE_TYPE_FANOUT)
+    # redeclaring an existing exchange with different attributes should fail
+    @test_throws AMQPClient.AMQPClientException exchange_declare(chan1, EXCG_FANOUT, EXCHANGE_TYPE_DIRECT)
+
+    # must reconnect as channel gets closed after a channel exception
+    close(chan1) # closing an already closed channel should be fine
+    chan1 = channel(conn, AMQPClient.UNUSED_CHANNEL, true)
+    @test chan1.id == 1
 
     # create and bind queues
     testlog("creating queues...")