throw when channel is closed due to server error

Some server errors lead to the channel being closed (server initiated). Certain API calls with invalid parameters may also trigger that (e.g. trying to redeclare an existing exchange as a different type). No response is sent for the failed API request in that case. Synchronous API calls that do not check for this condition would time out under this situation.

With this change, all synchronous API calls check for channel errors and throw an exception.

fixes #23

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...")