make sure we don't try to read past end of response body

Author: eli-darkly

lib/sse_client/streaming_http.rb

@@ -91,6 +91,7 @@ def initialize(socket, read_timeout)
       @read_timeout = read_timeout
       @parser = HTTPTools::Parser.new
       @buffer = ""
+      @done = false
       @lock = Mutex.new
 
       # Provide callbacks for the Parser to give us the headers and body. This has to be done
@@ -102,6 +103,9 @@ def initialize(socket, read_timeout)
       @parser.on(:stream) do |data|
         @lock.synchronize { @buffer << data }  # synchronize because we're called from another thread in Socketry
       end
+      @parser.on(:finish) do
+        @lock.synchronize { @done = true }
+      end
 
       # Block until the status code and headers have been successfully read.
       while !have_headers
@@ -132,6 +136,8 @@ def read_all
     # Attempt to read some more data from the socket. Return true if successful, false if EOF.
     # A read timeout will result in an exception from Socketry's readpartial method.
     def read_chunk_into_buffer
+      # If @done is set, it means the Parser has signaled end of response body
+      @lock.synchronize { return false if @done }
       data = @socket.readpartial(DEFAULT_CHUNK_SIZE, timeout: @read_timeout)
       return false if data == :eof
       @parser << data

spec/sse_client/streaming_http_spec.rb

@@ -73,6 +73,22 @@ def with_connection(cxn)
     end
   end
 
+  it "can read entire response body" do
+    body = <<-EOT
+This is
+a response
+EOT
+    with_server do |server|
+      server.setup_response("/foo") do |req,res|
+        res.body = body
+      end
+      with_connection(subject.new(server.base_uri.merge("/foo"), nil, {}, 30, 30)) do |cxn|
+        read_body = cxn.read_all
+        expect(read_body).to eq("This is\na response\n")
+      end
+    end
+  end
+
   it "enforces read timeout" do
     with_server do |server|
       server.setup_response("/") do |req,res|