ci: Add SSE contract test support (#25)

.github/actions/ci/action.yml

@@ -25,3 +25,8 @@ runs:
     - name: Run tests
       shell: bash
       run: bundle _2.2.10_ exec rspec spec
+
+    - name: Run contract tests
+      if: ${{ !startsWith(inputs.ruby-version, 'jruby') }}
+      shell: bash
+      run: make contract-tests

Makefile

@@ -0,0 +1,19 @@
+TEMP_TEST_OUTPUT=/tmp/sse-contract-test-service.log
+
+build-contract-tests:
+	@cd contract-tests && bundle _2.2.10_ install
+
+start-contract-test-service:
+	@cd contract-tests && bundle _2.2.10_ exec ruby service.rb
+
+start-contract-test-service-bg:
+	@echo "Test service output will be captured in $(TEMP_TEST_OUTPUT)"
+	@make start-contract-test-service >$(TEMP_TEST_OUTPUT) 2>&1 &
+
+run-contract-tests:
+	@curl -s https://raw.githubusercontent.com/launchdarkly/sse-contract-tests/v1.0.0/downloader/run.sh \
+      | VERSION=v1 PARAMS="-url http://localhost:8000 -debug -stop-service-at-end" sh
+
+contract-tests: build-contract-tests start-contract-test-service-bg run-contract-tests
+
+.PHONY: build-contract-tests start-contract-test-service run-contract-tests contract-tests

contract-tests/Gemfile

@@ -0,0 +1,10 @@
+source 'https://rubygems.org'
+
+gem 'ld-eventsource', path: '..'
+
+gem 'sinatra', '~> 2.1'
+# Sinatra can work with several server frameworks. In JRuby, we have to use glassfish (which
+# is only available in JRuby). Otherwise we use thin (which is not available in JRuby).
+gem 'glassfish', :platforms => :jruby
+gem 'thin', :platforms => :ruby
+gem 'json'

contract-tests/README.md

@@ -0,0 +1,5 @@
+# SSE client contract test service
+
+This directory contains an implementation of the cross-platform SSE testing protocol defined by https://github.com/launchdarkly/sse-contract-tests. See that project's `README` for details of this protocol, and the kinds of SSE client capabilities that are relevant to the contract tests. This code should not need to be updated unless the SSE client has added or removed such capabilities.
+
+To run these tests locally, run `make contract-tests` from the project root directory. This downloads the correct version of the test harness tool automatically.

contract-tests/service.rb

@@ -0,0 +1,77 @@
+require 'ld-eventsource'
+require 'json'
+require 'logger'
+require 'net/http'
+require 'sinatra'
+
+require './stream_entity.rb'
+
+$log = Logger.new(STDOUT)
+$log.formatter = proc {|severity, datetime, progname, msg|
+  "#{datetime.strftime('%Y-%m-%d %H:%M:%S.%3N')} #{severity} #{progname} #{msg}\n"
+}
+
+set :port, 8000
+set :logging, false
+
+streams = {}
+streamCounter = 0
+
+get '/' do
+  {
+    capabilities: [
+      'headers',
+      'last-event-id',
+      'read-timeout'
+    ]
+  }.to_json
+end
+
+delete '/' do
+  $log.info("Test service has told us to exit")
+  Thread.new { sleep 1; exit }
+  return 204
+end
+
+post '/' do
+  opts = JSON.parse(request.body.read, :symbolize_names => true)
+  streamUrl = opts[:streamUrl]
+  callbackUrl = opts[:callbackUrl]
+  tag = "[#{opts[:tag]}]:"
+
+  if !streamUrl || !callbackUrl
+    $log.error("#{tag} Received request with incomplete parameters: #{opts}")
+    return 400
+  end
+
+  streamCounter += 1
+  streamId = streamCounter.to_s
+  streamResourceUrl = "/streams/#{streamId}"
+
+  $log.info("#{tag} Starting stream from #{streamUrl}")
+  $log.debug("#{tag} Parameters: #{opts}")
+
+  entity = nil
+  sse = SSE::Client.new(
+    streamUrl,
+    headers: opts[:headers] || {},
+    last_event_id: opts[:lastEventId],
+    read_timeout: opts[:readTimeoutMs].nil? ? nil : (opts[:readTimeoutMs].to_f / 1000),
+    reconnect_time: opts[:initialDelayMs].nil? ? nil : (opts[:initialDelayMs].to_f / 1000)
+  ) do |sse|
+    entity = StreamEntity.new(sse, tag, callbackUrl)
+  end
+
+  streams[streamId] = entity
+
+  return [201, {"Location" => streamResourceUrl}, nil]
+end
+
+delete '/streams/:id' do |streamId|
+  entity = streams[streamId]
+  return 404 if entity.nil?
+  streams.delete(streamId)
+  entity.close
+
+  return 204
+end

contract-tests/stream_entity.rb

@@ -0,0 +1,58 @@
+require 'ld-eventsource'
+require 'json'
+require 'net/http'
+
+set :port, 8000
+set :logging, false
+
+class StreamEntity
+  def initialize(sse, tag, callbackUrl)
+    @sse = sse
+    @tag = tag
+    @callbackUrl = callbackUrl
+    @callbackCounter = 0
+
+    sse.on_event { |event| self.on_event(event) }
+    sse.on_error { |error| self.on_error(error) }
+  end
+
+  def on_event(event)
+    $log.info("#{@tag} Received event from stream (#{event.type})")
+    message = {
+      kind: 'event',
+      event: {
+        type: event.type,
+        data: event.data,
+        id: event.last_event_id
+      }
+    }
+    self.send_message(message)
+  end
+
+  def on_error(error)
+    $log.info("#{@tag} Received error from stream: #{error}")
+    message = {
+      kind: 'error',
+      error: error
+    }
+    self.send_message(message)
+  end
+
+  def send_message(message)
+    @callbackCounter += 1
+    uri = "#{@callbackUrl}/#{@callbackCounter}"
+    begin
+      resp = Net::HTTP.post(URI(uri), JSON.generate(message))
+      if resp.code.to_i >= 300
+        $log.error("#{@tag} Callback to #{url} returned status #{resp.code}")
+      end
+    rescue => e
+      $log.error("#{@tag} Callback to #{url} failed: #{e}")
+    end
+  end
+
+  def close
+    @sse.close
+    $log.info("#{@tag} Test ended")
+  end
+end