From 9f08d3730dc5199d615141e15e0c44040d8f285c Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 25 Sep 2026 11:31:08 -0500 Subject: [PATCH 1/3] feat!: Report a server-initiated stream close to the error handler When the server ends the stream and close was not called, on_error now gets an SSE::Errors::StreamClosedError. The client reconnects as before. This follows the WHATWG SSE processing model, where a normal end of body fires error before it reestablishes the connection. BREAKING CHANGE: A caller that calls close on any error now stops reconnecting after a server close. --- lib/ld-eventsource/client.rb | 5 +- lib/ld-eventsource/errors.rb | 10 ++++ spec/client_spec.rb | 101 +++++++++++++++++++++++++++++++++++ spec/headers_spec.rb | 5 +- 4 files changed, 116 insertions(+), 5 deletions(-) diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index bf6dbc8..c743d47 100644 --- a/lib/ld-eventsource/client.rb +++ b/lib/ld-eventsource/client.rb @@ -324,7 +324,10 @@ def run_stream # 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(resp) unless resp.nil? + unless resp.nil? + read_stream(resp) + log_and_dispatch_error(Errors::StreamClosedError.new, "Stream closed by server") unless @stopped.value + end rescue => e # When we deliberately close the connection, it will usually trigger an exception. The exact type # of exception depends on the specific Ruby runtime. But @stopped will always be set in this case. diff --git a/lib/ld-eventsource/errors.rb b/lib/ld-eventsource/errors.rb index 612c2da..1d0c449 100644 --- a/lib/ld-eventsource/errors.rb +++ b/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 StreamClosedError < StandardError + def initialize + super("stream closed by server") + end + end end end diff --git a/spec/client_spec.rb b/spec/client_spec.rb index 4a27223..fce5ab0 100644 --- a/spec/client_spec.rb +++ b/spec/client_spec.rb @@ -1416,4 +1416,105 @@ def test_object.to_s end end end + + describe "server-initiated stream close" do + it "passes one StreamClosedError 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 eq(SSE::Errors::StreamClosedError.new) + 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 "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 StreamClosedError, 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 diff --git a/spec/headers_spec.rb b/spec/headers_spec.rb index c0c3d77..c8e79ec 100644 --- a/spec/headers_spec.rb +++ b/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 = [] From 03c7b1a091f8438864dc373e8d50c20af6417d73 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 25 Sep 2026 11:42:10 -0500 Subject: [PATCH 2/3] fix: Raise the stream closed error where the end of the stream is read --- lib/ld-eventsource/client.rb | 9 +++------ spec/client_spec.rb | 34 +++++++++++++++++++++++++++++++++- 2 files changed, 36 insertions(+), 7 deletions(-) diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index c743d47..a8f344e 100644 --- a/lib/ld-eventsource/client.rb +++ b/lib/ld-eventsource/client.rb @@ -324,10 +324,7 @@ def run_stream # 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 - unless resp.nil? - read_stream(resp) - log_and_dispatch_error(Errors::StreamClosedError.new, "Stream closed by server") unless @stopped.value - end + read_stream(resp) unless resp.nil? rescue => e # When we deliberately close the connection, it will usually trigger an exception. The exact type # of exception depends on the specific Ruby runtime. But @stopped will always be set in this case. @@ -410,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::StreamClosedError end - break if data.nil? # keep for v5 compat + raise Errors::StreamClosedError if data.nil? # keep for v5 compat gen.yield data end end diff --git a/spec/client_spec.rb b/spec/client_spec.rb index fce5ab0..53ef426 100644 --- a/spec/client_spec.rb +++ b/spec/client_spec.rb @@ -1436,7 +1436,7 @@ def test_object.to_s with_client(client) do |c| expect(event_sink.pop).to eq(simple_event_1) - expect(error_sink.pop).to eq(SSE::Errors::StreamClosedError.new) + expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedError) expect(event_sink.pop).to eq(simple_event_2) expect(attempt).to eq 2 expect(error_sink.empty?).to be true @@ -1444,6 +1444,38 @@ def test_object.to_s end end + it "passes StreamClosedError 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::StreamClosedError) + 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| From 03536d462af679d07ce7fdaeed7ca09acdf0f280 Mon Sep 17 00:00:00 2001 From: jsonbailey Date: Fri, 25 Sep 2026 15:28:13 -0500 Subject: [PATCH 3/3] fix: Rename the stream closed error to StreamClosedByServerError --- lib/ld-eventsource/client.rb | 4 ++-- lib/ld-eventsource/errors.rb | 2 +- spec/client_spec.rb | 10 +++++----- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index a8f344e..936a6f9 100644 --- a/lib/ld-eventsource/client.rb +++ b/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 - raise Errors::StreamClosedError + raise Errors::StreamClosedByServerError end - raise Errors::StreamClosedError if data.nil? # keep for v5 compat + raise Errors::StreamClosedByServerError if data.nil? # keep for v5 compat gen.yield data end end diff --git a/lib/ld-eventsource/errors.rb b/lib/ld-eventsource/errors.rb index 1d0c449..f2b7b8d 100644 --- a/lib/ld-eventsource/errors.rb +++ b/lib/ld-eventsource/errors.rb @@ -84,7 +84,7 @@ def initialize(interval) # 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 StreamClosedError < StandardError + class StreamClosedByServerError < StandardError def initialize super("stream closed by server") end diff --git a/spec/client_spec.rb b/spec/client_spec.rb index 53ef426..e961950 100644 --- a/spec/client_spec.rb +++ b/spec/client_spec.rb @@ -1418,7 +1418,7 @@ def test_object.to_s end describe "server-initiated stream close" do - it "passes one StreamClosedError to the error handler, and the client reconnects" 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| @@ -1436,7 +1436,7 @@ def test_object.to_s with_client(client) do |c| expect(event_sink.pop).to eq(simple_event_1) - expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedError) + 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 @@ -1444,7 +1444,7 @@ def test_object.to_s end end - it "passes StreamClosedError when the HTTP library returns nil at the end of the stream" do + 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) @@ -1469,7 +1469,7 @@ def test_object.to_s with_client(client) do |c| expect(event_sink.pop).to eq(simple_event_1) - expect(error_sink.pop).to be_a(SSE::Errors::StreamClosedError) + 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 @@ -1498,7 +1498,7 @@ def test_object.to_s end end - it "passes ReadTimeoutError, not StreamClosedError, when the stream times out" do + it "passes ReadTimeoutError, not StreamClosedByServerError, when the stream times out" do with_server do |server| attempt = 0 server.setup_response("/") do |req,res|