diff --git a/lib/splitclient-rb/sse/event_source/client.rb b/lib/splitclient-rb/sse/event_source/client.rb index 14c33914..74c0cfc3 100644 --- a/lib/splitclient-rb/sse/event_source/client.rb +++ b/lib/splitclient-rb/sse/event_source/client.rb @@ -10,7 +10,7 @@ module SSE module EventSource class Client DEFAULT_READ_TIMEOUT = 70 - CONNECT_TIMEOUT = 30_000 + CONNECT_TIMEOUT = 30 OK_CODE = 200 KEEP_ALIVE_RESPONSE = "c\r\n:keepalive\n\n\r\n".freeze ERROR_EVENT_TYPE = 'error'.freeze @@ -94,7 +94,10 @@ def connect_stream(latch) partial_data = @socket.readpartial(10_000) read_first_event(partial_data, latch) - raise 'eof exception' if partial_data == :eof + if partial_data == :eof + @config.logger.error("SSE recived EOF unexpectedly") + return Constants::PUSH_RETRYABLE_ERROR + end rescue IO::WaitReadable => e @config.logger.debug("SSE client IO::WaitReadable transient error: #{e.inspect}") if @config.debug_enabled IO.select([@socket], nil, nil, @read_timeout) @@ -108,7 +111,7 @@ def connect_stream(latch) return Constants::PUSH_RETRYABLE_ERROR rescue EOFError => e @config.logger.error("SSE read operation EOF Exception!: #{e.inspect}") - raise 'eof exception' + return Constants::PUSH_RETRYABLE_ERROR rescue Errno::EBADF, IOError => e @config.logger.error("SSE read operation EBADF or IOError: #{e.inspect}") return Constants::PUSH_RETRYABLE_ERROR @@ -126,8 +129,6 @@ def connect_stream(latch) rescue Errno::EBADF @config.logger.debug("SSE socket is not connected (Errno::EBADF)") if @config.debug_enabled break - rescue RuntimeError - raise 'eof exception' rescue Exception => e @config.logger.debug("SSE socket is not connected: #{e.inspect}") if @config.debug_enabled break diff --git a/spec/sse/event_source/client_spec.rb b/spec/sse/event_source/client_spec.rb index e2665fa4..ca14cc47 100644 --- a/spec/sse/event_source/client_spec.rb +++ b/spec/sse/event_source/client_spec.rb @@ -285,10 +285,6 @@ sse_client.send(:connect_stream, latch) expect(log.string).to include 'SSE read operation timed out!' - allow(sse_client).to receive(:read_first_event).and_raise(EOFError) - expect { sse_client.send(:connect_stream, latch) }.to raise_error(RuntimeError) - expect(log.string).to include 'SSE read operation EOF Exception!' - allow(sse_client).to receive(:read_first_event).and_raise(Errno::EBADF) sse_client.send(:connect_stream, latch) expect(log.string).to include 'SSE read operation EBADF or IOError' @@ -305,6 +301,31 @@ end end + it 'test retry with EofError exceptions' do + mock_server do |server| + server.setup_response('/') do |_, res| + send_stream_content(res, event_occupancy) + end + start_workers + + sse_client = subject.new(config, api_token, telemetry_runtime_producer, event_parser, notification_manager_keeper, notification_processor, push_status_queue) + + sse_client.instance_variable_set(:@uri, URI(server.base_uri)) + latch = Concurrent::CountDownLatch.new(1) + + allow(sse_client).to receive(:read_first_event).and_raise(EOFError) + sleep(1) + thr1 = Thread.new do + sse_client.send(:connect_stream, latch) + end + sleep(1) + allow(sse_client).to receive(:read_first_event).and_return(true) + expect(log.string).to include 'SSE read operation EOF Exception' + + stop_workers + end + end + it 'test retry with EAGAIN exceptions' do mock_server do |server| server.setup_response('/') do |_, res|