From f3180058823f9ac7034c0e2914e121aa8a807d64 Mon Sep 17 00:00:00 2001
From: tan <tanmaykm@gmail.com>
Date: Mon, 25 Oct 2021 15:38:43 +0530
Subject: [PATCH] allow long polling request to be interrupted

Allow long polling request to be interrupted by closing the output channel.
There is no ready API/mechanism in Downloads.jl to interrupt a request.
Here we just throw an InterruptedException to the task that invoked Downloads.request.

Note: A better mechanism may be to invoke `remove_handle(multi::Multi, easy::Easy)`, but that should probably be through an interruption mechanism provided by Downloads.jl?
---
 src/client.jl                  | 31 ++++++++++++++++++++++++-------
 test/petstore/test_StoreApi.jl |  5 +++++
 2 files changed, 29 insertions(+), 7 deletions(-)

diff --git a/src/client.jl b/src/client.jl
index e4f67e3..8a0b10e 100644
--- a/src/client.jl
+++ b/src/client.jl
@@ -284,17 +284,20 @@ function do_request(ctx::Ctx, stream::Bool=false; stream_to::Union{Channel,Nothi
     try
         if stream
             @sync begin
-                @async begin
+                download_task = @async begin
                     try
                         resp = Downloads.request(resource_path;
                             input=input,
                             output=output,
                             kwargs...
                         )
-                        close(output)
                     catch ex
-                        @error("exception invoking request", exception=(ex,catch_backtrace()))
-                        rethrow()
+                        if !isa(ex, InterruptedException)
+                            @error("exception invoking request", exception=(ex,catch_backtrace()))
+                            rethrow()
+                        end
+                    finally
+                        close(output)
                     end
                 end
                 @async begin
@@ -304,10 +307,24 @@ function do_request(ctx::Ctx, stream::Bool=false; stream_to::Union{Channel,Nothi
                             data = response(return_type, resp, chunk)
                             put!(stream_to, data)
                         end
-                        close(stream_to)
                     catch ex
-                        @error("exception reading chunk", exception=(ex,catch_backtrace()))
-                        rethrow()
+                        if !isa(ex, InvalidStateException)
+                            @error("exception reading chunk", exception=(ex,catch_backtrace()))
+                            rethrow()
+                        end
+                    finally
+                        close(stream_to)
+                    end
+                end
+                @async begin
+                    while isopen(stream_to)
+                        try
+                            wait(stream_to)
+                            yield()
+                        catch ex
+                            isa(ex, InvalidStateException) || rethrow(ex)
+                            istaskdone(download_task) || schedule(download_task, InterruptException(), error=true)
+                        end
                     end
                 end
             end
diff --git a/test/petstore/test_StoreApi.jl b/test/petstore/test_StoreApi.jl
index 14a2b58..24627c4 100644
--- a/test/petstore/test_StoreApi.jl
+++ b/test/petstore/test_StoreApi.jl
@@ -42,6 +42,11 @@ function test(uri)
         end
     end
 
+    # a closed channel is equivalent of cancellation of the call, no error should be thrown
+    @test !isopen(response_channel)
+    resp = getOrderById(api, response_channel, 10)
+    @test (200 <= resp.status <= 206)
+
     @info("StoreApi - deleteOrder")
     @test deleteOrder(api, 10) === nothing