return retry when EOFError exception is caught

lib/splitclient-rb/sse/event_source/client.rb

@@ -10,7 +10,7 @@
     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
@@ -85,7 +85,7 @@
           end
         end
 
         def connect_stream(latch)
           return Constants::PUSH_RETRYABLE_ERROR unless socket_write(latch)
           while connected? || @first_event.value
             begin 
@@ -94,7 +94,10 @@
                   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 @@
                   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 @@
             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

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|