Simplify PR

Author: pankgeorg

src/subscriptions.jl

@@ -1,17 +1,5 @@
 const subscription_tracker = Ref{Dict}(Dict())
 
-function writews(ws::HTTP.WebSockets.WebSocket, msg)
-    if isdefined(HTTP, :send)
-        HTTP.send(ws, msg)
-    else
-        write(ws, msg)
-    end
-end
-
-function writews(ws::IO, msg)
-    write(ws, msg)
-end
-
 """
     open_subscription(fn::Function,
                       [client::Client],
@@ -109,7 +97,7 @@ function open_subscription(fn::Function,
     HTTP.WebSockets.open(client.ws_endpoint; retry=retry, headers=client.headers, suppress_close_error=false) do ws
         # Start sub
         output_info(verbose) && println("Starting $(get_name(subscription_name)) subscription with ID $sub_id")
-        writews(ws, message_str)
+        HTTP.send(ws, message_str)
         subscription_tracker[][sub_id] = "open"
 
         # Init function
@@ -170,24 +158,6 @@ function clear_subscriptions()
     end
 end
 
-function async_reader_with_timeout(io::IO, subtimeout)::Channel
-    ch = Channel(1)
-    task = @async begin
-        reader_task = current_task()
-        function timeout_cb(timer)
-            put!(ch, :timeout)
-            Base.throwto(reader_task, InterruptException())
-        end
-        timeout = Timer(timeout_cb, subtimeout)
-        data = readavailable(io)
-        subtimeout > 0 && close(timeout) # Cancel the timeout
-        put!(ch, data)
-    end
-    bind(ch, task)
-    return ch
-end
-
-
 function async_reader_with_timeout(ws::HTTP.WebSocket, subtimeout)::Channel
     ch = Channel(1)
     task = @async begin
@@ -205,28 +175,6 @@ function async_reader_with_timeout(ws::HTTP.WebSocket, subtimeout)::Channel
     return ch
 end
 
-
-function async_reader_with_stopfn(io::IO, stopfn, checktime)::Channel
-    ch = Channel(1) # Could we make this channel concretely typed?
-    task = @async begin
-        reader_task = current_task()
-        function timeout_cb(timer)
-            if stopfn()
-                put!(ch, :stopfn)
-                Base.throwto(reader_task, InterruptException())
-            else
-                timeout = Timer(timeout_cb, checktime)
-            end
-        end
-        timeout = Timer(timeout_cb, checktime)
-        data = readavailable(io)
-        close(timeout) # Cancel the timeout
-        put!(ch, data)
-    end
-    bind(ch, task)
-    return ch
-end
-
 function async_reader_with_stopfn(ws::HTTP.WebSockets.WebSocket, stopfn, checktime)::Channel
     ch = Channel(1) # Could we make this channel concretely typed?
     task = @async begin
@@ -265,20 +213,6 @@ A channel is returned with the data. If `stopfn` stops the websocket,
 the data will be `:stopfn`. If the timeout stops the websocket,
 the data will be `:timeout`
 """
-function readfromwebsocket(ws::IO, stopfn, subtimeout)
-    if isnothing(stopfn) && subtimeout > 0
-        ch_out = async_reader_with_timeout(ws, subtimeout)
-        data = take!(ch_out)
-    elseif !isnothing(stopfn)
-        checktime = subtimeout > 0 ? subtimeout : 2
-        ch_out = async_reader_with_stopfn(ws, stopfn, checktime)
-        data = take!(ch_out)
-    else
-        data = readavailable(ws)
-    end
-    return data
-end
-
 function readfromwebsocket(ws::HTTP.WebSockets.WebSocket, stopfn, subtimeout)
     if isnothing(stopfn) && subtimeout > 0
         ch_out = async_reader_with_timeout(ws, subtimeout)
@@ -291,4 +225,4 @@ function readfromwebsocket(ws::HTTP.WebSockets.WebSocket, stopfn, subtimeout)
         data = HTTP.receive(ws)
     end
     return data
-end
+end

test/subscriptions.jl

@@ -3,7 +3,7 @@ function listen_localhost()
         if HTTP.WebSockets.isupgrade(http.message)
             HTTP.WebSockets.upgrade(http) do ws
                 for data in ws
-                    GraphQLClient.writews(ws, data)
+                    HTTP.send(ws, data)
                 end
             end
         end
@@ -30,7 +30,7 @@ end
         @test take!(ch) == :timeout
 
         ch = GraphQLClient.async_reader_with_timeout(ws, 5)
-        GraphQLClient.writews(ws, "Data")
+        HTTP.send(ws, "Data")
         @test String(take!(ch)) == "Data"
 
         # stopfn
@@ -43,11 +43,11 @@ end
         @test take!(ch) == :stopfn
         stop[] = false
         ch = GraphQLClient.async_reader_with_stopfn(ws, stopfn, 0.5)
-        GraphQLClient.writews(ws, "Data")
+        HTTP.send(ws, "Data")
         @test String(take!(ch)) == "Data"
 
         # readfromwebsocket - no timeout or stopfn
-        GraphQLClient.writews(ws, "Data")
+        HTTP.send(ws, "Data")
         @test String(GraphQLClient.readfromwebsocket(ws, nothing, 0)) == "Data"
 
         # readfromwebsocket - timeout
@@ -90,7 +90,7 @@ function send_error_localhost(message, port)
                         }
                     }
                     """
-                    GraphQLClient.writews(ws, error_payload)
+                    HTTP.send(ws, error_payload)
                 end
             end
         end
@@ -116,7 +116,7 @@ function send_data_localhost(sub_name, port)
                         }
                     }
                     """
-                    GraphQLClient.writews(ws, data_payload)
+                    HTTP.send(ws, data_payload)
                 end
             end
         end
@@ -206,4 +206,4 @@ end
     @test isnothing(results[1].errors)
     @test !isnothing(results[1].data) # No point testing content as we've coded it in the test function
 
-end
+end