diff --git a/lib/ld-eventsource/client.rb b/lib/ld-eventsource/client.rb index bf6dbc8..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 - 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 diff --git a/lib/ld-eventsource/errors.rb b/lib/ld-eventsource/errors.rb index 612c2da..f2b7b8d 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 StreamClosedByServerError < 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..e961950 100644 --- a/spec/client_spec.rb +++ b/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 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 = []