prepare 5.0.1 release (#103)

lib/sse_client/sse_client.rb

@@ -61,6 +61,7 @@ def on_error(&action)
     def close
       if @stopped.make_true
         @cxn.close if !@cxn.nil?
+        @cxn = nil
       end
     end
 
@@ -77,14 +78,22 @@ def run_stream
         @cxn = nil
         begin
           @cxn = connect
+          # There's a potential race if close was called in the middle of the previous line, i.e. after we
+          # connected but before @cxn was set. Checking the variable again is a bit clunky but avoids that.
+          return if @stopped.value
           read_stream(@cxn) if !@cxn.nil?
         rescue Errno::EBADF
           # don't log this - it probably means we closed our own connection deliberately
         rescue StandardError => e
           @logger.error { "Unexpected error from event source: #{e.inspect}" }
           @logger.debug { "Exception trace: #{e.backtrace}" }
         end
-        @cxn.close if !cxn.nil?
+        begin
+          @cxn.close if !@cxn.nil?
+        rescue StandardError => e
+          @logger.error { "Unexpected error while closing stream: #{e.inspect}" }
+          @logger.debug { "Exception trace: #{e.backtrace}" }
+        end
       end
     end
 

lib/sse_client/streaming_http.rb

@@ -1,3 +1,4 @@
+require "concurrent/atomics"
 require "http_tools"
 require "socketry"
 
@@ -15,11 +16,14 @@ def initialize(uri, proxy, headers, connect_timeout, read_timeout)
       @reader = HTTPResponseReader.new(@socket, read_timeout)
       @status = @reader.status
       @headers = @reader.headers
+      @closed = Concurrent::AtomicBoolean.new(false)
     end
 
     def close
-      @socket.close if @socket
-      @socket = nil
+      if @closed.make_true
+        @socket.close if @socket
+        @socket = nil
+      end
     end
 
     # Generator that returns one line of the response body at a time (delimited by \r, \n,

spec/sse_client/sse_client_spec.rb

@@ -136,4 +136,42 @@ def with_client(client)
       end
     end
   end
+
+  it "reconnects if stream returns EOF" do
+    events_body_1 = <<-EOT
+event: go
+data: foo
+
+EOT
+    events_body_2 = <<-EOT
+event: go
+data: bar
+
+EOT
+    with_server do |server|
+      attempt = 0
+      server.setup_response("/") do |req,res|
+        attempt += 1
+        if attempt == 1
+          res.body = events_body_1
+        else
+          res.body = events_body_2
+        end
+        res.content_type = "text/event-stream"
+        res.status = 200
+      end
+
+      event_sink = Queue.new
+      client = subject.new(server.base_uri,
+            reconnect_time: 0.25, read_timeout: 0.25) do |c|
+        c.on_event { |event| event_sink << event }
+      end
+
+      with_client(client) do |client|
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:go, "foo", nil))
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:go, "bar", nil))
+        expect(attempt).to be >= 2
+      end
+    end
+  end
 end