LaunchDarkly.EventSource 6.0 rewrite, part 1

.circleci/config.yml

@@ -34,7 +34,7 @@ jobs:
     steps:
       - run:
           name: install packages
-          command: apt-get -q update && apt-get install -qy awscli
+          command: apt-get -q update && apt-get install -qy make
       - checkout
       - run:
           name: build
@@ -48,6 +48,12 @@ jobs:
       - store_test_results:
           path: /tmp/circle-reports
 
+      - run: make build-contract-tests
+      - run:
+          command: make start-contract-test-service
+          background: true
+      - run: make run-contract-tests
+
   test_windows:
     parameters:
       test-target-framework:

.gitignore

@@ -40,6 +40,7 @@ local.properties
 *.suo
 *.user
 *.sln.docstates
+launchSettings.json
 
 # Build results
 [Dd]ebug/

.ldrelease/config.yml

@@ -21,4 +21,4 @@ jobs:
 
 documentation:
   gitHubPages: true
-  title: LaunchDarkly .NET EventSource Client
+  title: LaunchDarkly .NET SSE Client

CONTRIBUTING.md

@@ -50,4 +50,9 @@ Or, to run tests only for the .NET Standard 2.0 target (using the .NET Core 3.1
 dotnet test test/LaunchDarkly.EventSource.Tests -f netcoreapp3.1
 ```
 
+To run the standardized contract tests that are run against all LaunchDarkly SSE client implementations (this is currently not supported on Windows):
+```
+make contract-tests
+```
+
 Note that the unit tests can only be run in Debug configuration. There is an `InternalsVisibleTo` directive that allows the test code to access internal members of the library, and assembly strong-naming in the Release configuration interferes with this.

Makefile

@@ -0,0 +1,30 @@
+
+build:
+	dotnet build
+
+test:
+	dotnet test
+
+clean:
+	dotnet clean
+
+TEMP_TEST_OUTPUT=/tmp/sse-contract-test-service.log
+TESTFRAMEWORK ?= netcoreapp3.1
+
+build-contract-tests:
+	@cd contract-tests && BUILDFRAMEWORK=netstandard2.0 dotnet build TestService.csproj
+
+start-contract-test-service:
+	@cd contract-tests && dotnet bin/Debug/${TESTFRAMEWORK}/ContractTestService.dll
+
+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 test clean build-contract-tests start-contract-test-service run-contract-tests contract-tests

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/Representations.cs

@@ -0,0 +1,43 @@
+using System.Collections.Generic;
+using System.Text.Json.Serialization;
+
+namespace TestService
+{
+    public class Status
+    {
+        [JsonPropertyName("capabilities")] public string[] Capabilities { get; set; }
+    }
+
+    public class StreamOptions
+    {
+        [JsonPropertyName("streamUrl")] public string StreamUrl { get; set; }
+        [JsonPropertyName("callbackUrl")] public string CallbackUrl { get; set; }
+        [JsonPropertyName("tag")] public string Tag { get; set; }
+        [JsonPropertyName("headers")] public Dictionary<string, string> Headers { get; set; }
+        [JsonPropertyName("initialDelayMs")] public int? InitialDelayMs { get; set; }
+        [JsonPropertyName("readTimeoutMs")] public int? ReadTimeoutMs { get; set; }
+        [JsonPropertyName("lastEventId")] public string LastEventId { get; set; }
+        [JsonPropertyName("method")] public string Method { get; set; }
+        [JsonPropertyName("body")] public string Body { get; set; }
+    }
+
+    public class CallbackMessage
+    {
+        [JsonPropertyName("kind")] public string Kind { get; set; }
+        [JsonPropertyName("event")] public CallbackEventMessage Event { get; set; }
+        [JsonPropertyName("comment")] public string Comment { get; set; }
+        [JsonPropertyName("error")] public string Error { get; set; }
+    }
+
+    public class CallbackEventMessage
+    {
+        [JsonPropertyName("type")] public string Type { get; set; }
+        [JsonPropertyName("data")] public string Data { get; set; }
+        [JsonPropertyName("id")] public string Id { get; set; }
+    }
+
+    public class CommandParams
+    {
+        [JsonPropertyName("command")] public string Command { get; set; }
+    }
+}

contract-tests/StreamEntity.cs

@@ -0,0 +1,220 @@
+using System;
+using System.Net.Http;
+using System.Text;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Xml.Linq;
+using LaunchDarkly.EventSource;
+using LaunchDarkly.EventSource.Events;
+using LaunchDarkly.EventSource.Exceptions;
+using LaunchDarkly.Logging;
+
+namespace TestService
+{
+    public class StreamEntity
+    {
+        private static HttpClient _httpClient = new HttpClient();
+
+        private readonly EventSource _stream;
+        private readonly StreamOptions _options;
+        private readonly Logger _log;
+        private volatile int _callbackMessageCounter;
+        private volatile bool _closed;
+
+        public StreamEntity(
+            StreamOptions options,
+            Logger log
+        )
+        {
+            _options = options;
+            _log = log;
+
+            var httpConfig = ConnectStrategy.Http(new Uri(options.StreamUrl));
+            if (_options.Headers != null)
+            {
+                foreach (var kv in _options.Headers)
+                {
+                    if (kv.Key.ToLower() != "content-type")
+                    {
+
+                        httpConfig = httpConfig.Header(kv.Key, kv.Value);
+                    }
+                }
+            }
+            if (_options.Method != null)
+            {
+                httpConfig = httpConfig.Method(new HttpMethod(_options.Method));
+                var contentType = "text/plain";
+                if (_options.Headers != null)
+                {
+                    foreach (var kv in _options.Headers)
+                    {
+                        if (kv.Key.ToLower() == "content-type")
+                        {
+                            contentType = kv.Value;
+                            if (contentType.Contains(";"))
+                            {
+                                contentType = contentType.Substring(0, contentType.IndexOf(";"));
+                            }
+                        }
+                    }
+                }
+                httpConfig = httpConfig.RequestBody(_options.Body, contentType);
+            }
+            if (_options.ReadTimeoutMs != null)
+            {
+                httpConfig = httpConfig.ReadTimeout(TimeSpan.FromMilliseconds(_options.ReadTimeoutMs.Value));
+            }
+
+            var builder = Configuration.Builder(httpConfig)
+                .ErrorStrategy(ErrorStrategy.AlwaysContinue) // see comments in RunAsync
+                .Logger(log);
+
+            if (_options.InitialDelayMs != null)
+            {
+                builder.InitialRetryDelay(TimeSpan.FromMilliseconds(_options.InitialDelayMs.Value));
+            }
+            if (_options.LastEventId != null)
+            {
+                builder.LastEventId(_options.LastEventId);
+            }
+
+            _log.Info("Opening stream from {0}", _options.StreamUrl);
+
+            _stream = new EventSource(builder.Build());
+
+            Task.Run(RunAsync);
+        }
+
+        private async Task RunAsync()
+        {
+            // A typical SSE-based application would only be interested in the events that
+            // are represented by EventMessage, so it could more simply call ReadMessageAsync()
+            // and receive only that type. But for the contract tests, we want to know more
+            // details such as comments and error conditions.
+
+            while (_stream.ReadyState != ReadyState.Shutdown)
+            {
+                try
+                {
+                    var e = await _stream.ReadAnyEventAsync();
+
+                    if (e is MessageEvent me)
+                    {
+                        _log.Info("Received event from stream (type: {0}, data: {1})",
+                            me.Name, me.Data);
+                        await SendMessage(new CallbackMessage
+                        {
+                            Kind = "event",
+                            Event = new CallbackEventMessage
+                            {
+                                Type = me.Name,
+                                Data = me.Data,
+                                Id = me.LastEventId
+                            }
+                        });
+                    }
+                    else if (e is CommentEvent ce)
+                    {
+                        _log.Info("Received comment from stream: {0}", ce.Text);
+                        await SendMessage(new CallbackMessage
+                        {
+                            Kind = "comment",
+                            Comment = ce.Text
+                        });
+                    }
+                    else if (e is FaultEvent fe)
+                    {
+                        if (fe.Exception is StreamClosedByCallerException)
+                        {
+                            // This one is special because it simply means we deliberately
+                            // closed the stream ourselves, so we don't need to report it
+                            // to the test harness.
+                            continue;
+                        }
+                        var exDesc = LogValues.ExceptionSummary(fe.Exception).ToString();
+                        _log.Info("Received error from stream: {0}", exDesc);
+                        await SendMessage(new CallbackMessage
+                        {
+                            Kind = "error",
+                            Error = exDesc
+                        });
+                    }
+                }
+                catch (Exception ex)
+                {
+                    // Because we specified ErrorStrategy.AlwaysContinue, EventSource
+                    // will normally report any errors on the stream as just part of the
+                    // stream, so we will get them as FaultEvents and then it will
+                    // transparently reconnect. Any exception that is thrown here
+                    // probably means there is a bug, so we'll report it to the test
+                    // harness (likely causing test to fail) and also log a detailed
+                    // stacktrace.
+                    _log.Error("Unexpected exception: {0} {1}", LogValues.ExceptionSummary(ex),
+                        LogValues.ExceptionTrace(ex));
+                    await SendMessage(new CallbackMessage
+                    {
+                        Kind = "error",
+                        Error = LogValues.ExceptionSummary(ex).ToString()
+                    });
+                }
+            }
+        }
+
+        public void Close()
+        {
+            _closed = true;
+            _log.Info("Closing");
+            _stream.Close();
+            _log.Info("Test ended");
+        }
+
+        public bool DoCommand(string command)
+        {
+            _log.Info("Test harness sent command: {0}", command);
+            if (command == "restart")
+            {
+                _stream.Interrupt();
+                return true;
+            }
+            return false;
+        }
+
+        private async Task SendMessage(object message)
+        {
+            if (_closed)
+            {
+                return;
+            }
+            var json = JsonSerializer.Serialize(message);
+            _log.Info("Sending: {0}", json);
+            var counter = Interlocked.Increment(ref _callbackMessageCounter);
+            var uri = new Uri(_options.CallbackUrl + "/" + counter);
+
+            using (var request = new HttpRequestMessage(HttpMethod.Post, uri))
+            using (var stringContent = new StringContent(json, Encoding.UTF8, "application/json"))
+            {
+                request.Content = stringContent;
+                try
+                {
+                    using (var response = await _httpClient.SendAsync(request))
+                    {
+                        if (!response.IsSuccessStatusCode)
+                        {
+                            _log.Error("Callback to {0} returned HTTP {1}", uri, response.StatusCode);
+                        }
+                        else
+                        {
+                            _log.Info("Callback to {0} succeeded", uri);
+                        }
+                    }
+                }
+                catch (Exception e)
+                {
+                    _log.Error("Callback to {0} failed: {1}", uri, e.GetType());
+                }
+            }
+        }
+    }
+}