Merge pull request #69 from launchdarkly/eb/ch19217/http-tests

eli-darkly · web-flow · commit c66a8e62ea25 · 2018-06-22T12:35:17.000-07:00
end-to-end tests for new SSE client [2 of 2]
diff --git a/lib/sse_client/sse_client.rb b/lib/sse_client/sse_client.rb
@@ -19,7 +19,7 @@ def initialize(uri, options = {})
       @uri = URI(uri)
       @stopped = Concurrent::AtomicBoolean.new(false)
 
-      @headers = options[:headers].clone || {}
+      @headers = options[:headers] ? options[:headers].clone : {}
       @connect_timeout = options[:connect_timeout] || DEFAULT_CONNECT_TIMEOUT
       @read_timeout = options[:read_timeout] || DEFAULT_READ_TIMEOUT
       @logger = options[:logger] || default_logger
@@ -40,7 +40,7 @@ def initialize(uri, options = {})
 
       yield self if block_given?
 
-      @worker = Thread.new do
+      Thread.new do
         run_stream
       end
     end
@@ -59,7 +59,7 @@ def on_error(&action)
 
     def close
       if @stopped.make_true
-        @worker.raise ShutdownSignal.new
+        @cxn.close if !@cxn.nil?
       end
     end
 
@@ -73,23 +73,24 @@ def default_logger
 
     def run_stream
       while !@stopped.value
-        cxn = nil
+        @cxn = nil
         begin
-          cxn = connect
-          read_stream(cxn)
-        rescue ShutdownSignal
-          return
+          @cxn = connect
+          read_stream(@cxn)
+        rescue Errno::EBADF
+          # don't log this - it probably means we closed our own connection deliberately
         rescue StandardError => e
           @logger.error { "Unexpected error from event source: #{e.inspect}" }
           @logger.debug { "Exception trace: #{e.backtrace}" }
         end
-        cxn.close if !cxn.nil?
+        @cxn.close if !cxn.nil?
       end
     end
 
     # Try to establish a streaming connection. Returns the StreamingHTTPConnection object if successful.
     def connect
       loop do
+        return if @stopped.value
         interval = @backoff.next_interval
         if interval > 0
           @logger.warn { "Will retry connection after #{'%.3f' % interval} seconds" } 
@@ -101,10 +102,13 @@ def connect
             body = cxn.read_all  # grab the whole response body in case it has error details
             cxn.close
             @on[:error].call({status_code: cxn.status, body: body})
+            next
           elsif cxn.headers["content-type"] && cxn.headers["content-type"].start_with?("text/event-stream")
             return cxn  # we're good to proceed
           end
           @logger.error { "Event source returned unexpected content type '#{cxn.headers["content-type"]}'" }
+        rescue Errno::EBADF
+          raise
         rescue StandardError => e
           @logger.error { "Unexpected error from event source: #{e.inspect}" }
           @logger.debug { "Exception trace: #{e.backtrace}" }
@@ -124,6 +128,7 @@ def open_connection(headers)
     def read_stream(cxn)
       event_parser = EventParser.new(cxn.read_lines)
       event_parser.items.each do |item|
+        return if @stopped.value
         case item
           when SSEEvent
             dispatch_event(item)
@@ -154,8 +159,4 @@ def build_headers
       h.merge(@headers)
     end
   end
-
-  # Custom exception that we use to tell the worker thread to stop
-  class ShutdownSignal < StandardError
-  end
 end
diff --git a/lib/sse_client/streaming_http.rb b/lib/sse_client/streaming_http.rb
@@ -91,6 +91,7 @@ def initialize(socket, read_timeout)
       @read_timeout = read_timeout
       @parser = HTTPTools::Parser.new
       @buffer = ""
+      @done = false
       @lock = Mutex.new
 
       # Provide callbacks for the Parser to give us the headers and body. This has to be done
@@ -102,6 +103,9 @@ def initialize(socket, read_timeout)
       @parser.on(:stream) do |data|
         @lock.synchronize { @buffer << data }  # synchronize because we're called from another thread in Socketry
       end
+      @parser.on(:finish) do
+        @lock.synchronize { @done = true }
+      end
 
       # Block until the status code and headers have been successfully read.
       while !have_headers
@@ -132,6 +136,8 @@ def read_all
     # Attempt to read some more data from the socket. Return true if successful, false if EOF.
     # A read timeout will result in an exception from Socketry's readpartial method.
     def read_chunk_into_buffer
+      # If @done is set, it means the Parser has signaled end of response body
+      @lock.synchronize { return false if @done }
       data = @socket.readpartial(DEFAULT_CHUNK_SIZE, timeout: @read_timeout)
       return false if data == :eof
       @parser << data
diff --git a/spec/sse_client/sse_client_spec.rb b/spec/sse_client/sse_client_spec.rb
@@ -0,0 +1,139 @@
+require "spec_helper"
+require "socketry"
+require "sse_client/sse_shared"
+
+#
+# End-to-end tests of SSEClient against a real server
+#
+describe SSE::SSEClient do
+  subject { SSE::SSEClient }
+
+  def with_client(client)
+    begin
+      yield client
+    ensure
+      client.close
+    end
+  end
+
+  it "sends expected headers" do
+    with_server do |server|
+      requests = Queue.new
+      server.setup_response("/") do |req,res|
+        requests << req
+        res.content_type = "text/event-stream"
+        res.status = 200
+      end
+      
+      headers = {
+        "Authorization" => "secret"
+      }
+
+      with_client(subject.new(server.base_uri, headers: headers)) do |client|
+        received_req = requests.pop
+        expect(received_req.header).to eq({
+          "accept" => ["text/event-stream"],
+          "cache-control" => ["no-cache"],
+          "host" => ["127.0.0.1"],
+          "authorization" => ["secret"]
+        })
+      end
+    end
+  end
+
+  it "receives messages" do
+    events_body = <<-EOT
+event: go
+data: foo
+id: 1
+
+event: stop
+data: bar
+
+EOT
+    with_server do |server|
+      server.setup_response("/") do |req,res|
+        res.content_type = "text/event-stream"
+        res.status = 200
+        res.body = events_body
+      end
+
+      event_sink = Queue.new
+      client = subject.new(server.base_uri) do |c|
+        c.on_event { |event| event_sink << event }
+      end
+
+      with_client(client) do |client|
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:go, "foo", "1"))
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:stop, "bar", nil))
+      end
+    end
+  end
+
+  it "reconnects after error response" do
+    events_body = <<-EOT
+event: go
+data: foo
+
+EOT
+    with_server do |server|
+      attempt = 0
+      server.setup_response("/") do |req,res|
+        attempt += 1
+        if attempt == 1
+          res.status = 500
+          res.body = "sorry"
+          res.keep_alive = false
+        else
+          res.content_type = "text/event-stream"
+          res.status = 200
+          res.body = events_body
+        end
+      end
+
+      event_sink = Queue.new
+      error_sink = Queue.new
+      client = subject.new(server.base_uri, reconnect_time: 0.25) do |c|
+        c.on_event { |event| event_sink << event }
+        c.on_error { |error| error_sink << error }
+      end
+
+      with_client(client) do |client|
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:go, "foo", nil))
+        expect(error_sink.pop).to eq({ status_code: 500, body: "sorry" })
+        expect(attempt).to be >= 2
+      end
+    end
+  end
+
+  it "reconnects after read timeout" do
+    events_body = <<-EOT
+event: go
+data: foo
+
+EOT
+    with_server do |server|
+      attempt = 0
+      server.setup_response("/") do |req,res|
+        attempt += 1
+        if attempt == 1
+          sleep(2)
+        end
+        res.content_type = "text/event-stream"
+        res.status = 200
+        res.body = events_body
+      end
+
+      event_sink = Queue.new
+      client = subject.new(server.base_uri,
+          reconnect_time: 0.25, read_timeout: 0.25) do |c|
+        c.on_event { |event| event_sink << event }
+      end
+
+      with_client(client) do |client|
+        expect(event_sink.pop).to eq(SSE::SSEEvent.new(:go, "foo", nil))
+        expect(attempt).to be >= 2
+      end
+    end
+  end
+end
diff --git a/spec/sse_client/sse_shared.rb b/spec/sse_client/sse_shared.rb
@@ -0,0 +1,51 @@
+require "spec_helper"
+require "webrick"
+
+class StubHTTPServer
+  def initialize
+    @port = 50000
+    begin
+      @server = WEBrick::HTTPServer.new(
+        BindAddress: '127.0.0.1',
+        Port: @port,
+        AccessLog: [],
+        Logger: NullLogger.new
+      )
+    rescue Errno::EADDRINUSE
+      @port += 1
+      retry
+    end
+  end
+
+  def start
+    Thread.new { @server.start }
+  end
+
+  def stop
+    @server.shutdown
+  end
+
+  def base_uri
+    URI("http://127.0.0.1:#{@port}")
+  end
+
+  def setup_response(uri_path, &action)
+    @server.mount_proc(uri_path, action)
+  end
+end
+
+class NullLogger
+  def method_missing(*)
+    self
+  end
+end
+
+def with_server
+  server = StubHTTPServer.new
+  begin
+    server.start
+    yield server
+  ensure
+    server.stop
+  end
+end
diff --git a/spec/sse_client/streaming_http_spec.rb b/spec/sse_client/streaming_http_spec.rb
