feat!: Report a server-initiated stream close to the error handler

lib/ld-eventsource/client.rb

@@ -407,9 +407,9 @@ def read_stream(cxn)
               # For historical reasons, we rethrow this as our own type
               raise Errors::ReadTimeoutError.new(@read_timeout)
             rescue EOFError
-              break
+              raise Errors::StreamClosedByServerError
             end
-            break if data.nil?  # keep for v5 compat
+            raise Errors::StreamClosedByServerError if data.nil?  # keep for v5 compat
             gen.yield data
           end
         end

lib/ld-eventsource/errors.rb

@@ -79,5 +79,15 @@ def initialize(interval)
         super("no data received in #{interval} seconds")
       end
     end
+
+    #
+    # An exception class indicating that the server ended the stream without an error. This is
+    # passed to the error handler specified in {Client#on_error}, and the client then reconnects.
+    #
+    class StreamClosedByServerError < StandardError
+      def initialize
+        super("stream closed by server")
+      end
+    end
   end
 end

spec/client_spec.rb

@@ -1416,4 +1416,137 @@ def test_object.to_s
       end
     end
   end
+
+  describe "server-initiated stream close" do
+    it "passes one StreamClosedByServerError to the error handler, and the client reconnects" do
+      with_server do |server|
+        attempt = 0
+        server.setup_response("/") do |req,res|
+          attempt += 1
+          send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
+            keep_open: attempt == 2)
+        end
+
+        event_sink = Queue.new
+        error_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| error_sink << error }
+        end
+
+        with_client(client) do |c|
+          expect(event_sink.pop).to eq(simple_event_1)
+          expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedByServerError)
+          expect(event_sink.pop).to eq(simple_event_2)
+          expect(attempt).to eq 2
+          expect(error_sink.empty?).to be true
+        end
+      end
+    end
+
+    it "passes StreamClosedByServerError when the HTTP library returns nil at the end of the stream" do
+      # http 4 and 5 return nil from readpartial at the end of the stream; http 6 raises EOFError.
+      allow_any_instance_of(HTTP::Response).to receive(:readpartial).and_wrap_original do |m, *args|
+        m.call(*args)
+      rescue EOFError
+        nil
+      end
+
+      with_server do |server|
+        attempt = 0
+        server.setup_response("/") do |req,res|
+          attempt += 1
+          send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
+            keep_open: attempt == 2)
+        end
+
+        event_sink = Queue.new
+        error_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| error_sink << error }
+        end
+
+        with_client(client) do |c|
+          expect(event_sink.pop).to eq(simple_event_1)
+          expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedByServerError)
+          expect(event_sink.pop).to eq(simple_event_2)
+          expect(attempt).to eq 2
+        end
+      end
+    end
+
+    it "does not pass an error to the error handler when the client is closed" do
+      with_server do |server|
+        server.setup_response("/") do |req,res|
+          send_stream_content(res, simple_event_1_text, keep_open: true)
+        end
+
+        event_sink = Queue.new
+        error_sink = Queue.new
+        client = subject.new(server.base_uri) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| error_sink << error }
+        end
+
+        with_client(client) do |c|
+          event_sink.pop  # wait till we have definitely started reading the stream
+          c.close
+          sleep 0.25  # there's no way to really know when the stream thread has finished
+          expect(error_sink.empty?).to be true
+        end
+      end
+    end
+
+    it "passes ReadTimeoutError, not StreamClosedByServerError, when the stream times out" do
+      with_server do |server|
+        attempt = 0
+        server.setup_response("/") do |req,res|
+          attempt += 1
+          send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
+            keep_open: true)
+        end
+
+        event_sink = Queue.new
+        error_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap, read_timeout: 0.25) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| error_sink << error }
+        end
+
+        with_client(client) do |c|
+          expect(event_sink.pop).to eq(simple_event_1)
+          expect(event_sink.pop).to eq(simple_event_2)
+          c.close
+          errors = []
+          errors << error_sink.pop until error_sink.empty?
+          expect(errors).not_to be_empty
+          expect(errors).to all(be_a(SSE::Errors::ReadTimeoutError))
+        end
+      end
+    end
+
+    it "reconnects if the error handler raises an exception" do
+      with_server do |server|
+        attempt = 0
+        server.setup_response("/") do |req,res|
+          attempt += 1
+          send_stream_content(res, attempt == 1 ? simple_event_1_text : simple_event_2_text,
+            keep_open: attempt == 2)
+        end
+
+        event_sink = Queue.new
+        client = subject.new(server.base_uri, reconnect_time: reconnect_asap) do |c|
+          c.on_event { |event| event_sink << event }
+          c.on_error { |error| raise "handler failure" }
+        end
+
+        with_client(client) do |c|
+          expect(event_sink.pop).to eq(simple_event_1)
+          expect(event_sink.pop).to eq(simple_event_2)
+          expect(attempt).to eq 2
+        end
+      end
+    end
+  end
 end

spec/headers_spec.rb

@@ -364,11 +364,8 @@ def send_stream_content(res, content, keep_open:)
     it "handles all three callback types without conflicts" do
       with_server do |server|
         server.setup_response("/") do |req,res|
-          res.status = 200
-          res.content_type = "text/event-stream"
           res['X-Test'] = 'multi-callback'
-          res.chunked = true
-          res.body = proc { |out| out.write("data: message\n\n") }
+          send_stream_content(res, "data: message\n\n", keep_open: true)
         end
 
         callbacks_fired = []