fix: SSE streaming stall on MAUI Android (SDK-2755)

contract-tests/README.md

@@ -0,0 +1,32 @@
+# SSE contract-tests service
+
+This is a small HTTP service that wraps `LaunchDarkly.EventSource` and exposes it to
+the [`sse-contract-tests`](https://github.com/launchdarkly/sse-contract-tests) harness.
+It is the .NET analogue of the `contract-tests/` services in
+[`eventsource`](https://github.com/launchdarkly/eventsource) (Go),
+[`js-eventsource`](https://github.com/launchdarkly/js-eventsource) (JavaScript), and
+[`okhttp-eventsource`](https://github.com/launchdarkly/okhttp-eventsource) (Java).
+
+The service is not shipped as part of any published NuGet package. It exists purely
+as a test target for the `sse-contract-tests` harness.
+
+## Running
+
+```
+dotnet run --project TestService.csproj
+```
+
+The service listens on port 8000 by default.
+
+## Running the harness against it
+
+Either use the released harness binary via `sse-contract-tests`'s `downloader/run.sh`
+script, or build the harness locally:
+
+```
+cd path/to/sse-contract-tests
+go build -o sse-test-harness .
+./sse-test-harness --url http://localhost:8000
+```
+
+To exercise only a subset of tests, pass `--run <pattern>` or `--skip <pattern>`.

contract-tests/Representations.cs

@@ -0,0 +1,46 @@
+using System.Collections.Generic;
+
+// Note, in order for System.Text.Json serialization/deserialization to work correctly, the members of
+// these classes must be properties with get/set, rather than fields. The property names are automatically
+// camelCased by System.Text.Json.
+
+namespace TestService
+{
+    public class Status
+    {
+        public string[] Capabilities { get; set; }
+    }
+
+    public class StreamOptions
+    {
+        public string StreamUrl { get; set; }
+        public string CallbackUrl { get; set; }
+        public string Tag { get; set; }
+        public Dictionary<string, string> Headers { get; set; }
+        public int? InitialDelayMs { get; set; }
+        public int? ReadTimeoutMs { get; set; }
+        public string LastEventId { get; set; }
+        public string Method { get; set; }
+        public string Body { get; set; }
+    }
+
+    public class Message
+    {
+        public string Kind { get; set; }
+        public EventMessage Event { get; set; }
+        public string Comment { get; set; }
+        public string Error { get; set; }
+    }
+
+    public class EventMessage
+    {
+        public string Type { get; set; }
+        public string Data { get; set; }
+        public string Id { get; set; }
+    }
+
+    public class CommandParams
+    {
+        public string Command { get; set; }
+    }
+}

contract-tests/StreamEntity.cs

@@ -0,0 +1,173 @@
+using System;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Text;
+using System.Text.Json;
+using System.Threading;
+using LaunchDarkly.EventSource;
+using LaunchDarkly.Logging;
+
+namespace TestService
+{
+    /// <summary>
+    /// Wraps a single LaunchDarkly.EventSource.EventSource instance driven by the SSE contract-tests
+    /// harness. Subscribes to the SSE client's events and forwards each one to the harness callback
+    /// URL as a JSON message.
+    /// </summary>
+    /// <remarks>
+    /// This is the .NET analogue of ssetest.StreamEntity in okhttp-eventsource's contract-tests
+    /// service. The main structural difference: .NET's EventSource is push-based (fires event
+    /// handlers on its own internal thread), so we don't need a background reader thread the way
+    /// the Java service does around its blocking event iterator.
+    /// </remarks>
+    public class StreamEntity
+    {
+        private readonly StreamOptions _options;
+        private readonly Logger _logger;
+        private readonly HttpClient _callbackClient;
+        private readonly EventSource _eventSource;
+        private int _callbackMessageCounter;
+        private volatile bool _closed;
+
+        private static readonly JsonSerializerOptions JsonOptions = new JsonSerializerOptions
+        {
+            PropertyNamingPolicy = JsonNamingPolicy.CamelCase,
+            DefaultIgnoreCondition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull,
+        };
+
+        public StreamEntity(StreamOptions options, ILogAdapter logAdapter)
+        {
+            _options = options;
+            _logger = logAdapter.Logger(options.Tag ?? "stream");
+            _callbackClient = new HttpClient();
+
+            _logger.Info("Opening stream to {0}", options.StreamUrl);
+
+            var configBuilder = Configuration.Builder(new Uri(options.StreamUrl));
+
+            if (options.Headers != null)
+            {
+                foreach (var kv in options.Headers)
+                {
+                    // Content-Type is a content header, not a request header; .NET's HttpHeaders
+                    // throws if we try to add it as a request header. We consume this header
+                    // separately below when constructing the request body.
+                    if (string.Equals(kv.Key, "content-type", System.StringComparison.OrdinalIgnoreCase))
+                    {
+                        continue;
+                    }
+                    configBuilder.RequestHeader(kv.Key, kv.Value);
+                }
+            }
+            if (options.InitialDelayMs.HasValue)
+            {
+                configBuilder.InitialRetryDelay(TimeSpan.FromMilliseconds(options.InitialDelayMs.Value));
+            }
+            if (options.ReadTimeoutMs.HasValue)
+            {
+                configBuilder.ReadTimeout(TimeSpan.FromMilliseconds(options.ReadTimeoutMs.Value));
+            }
+            if (!string.IsNullOrEmpty(options.LastEventId))
+            {
+                configBuilder.LastEventId(options.LastEventId);
+            }
+            if (!string.IsNullOrEmpty(options.Method))
+            {
+                configBuilder.Method(new HttpMethod(options.Method));
+                if (!string.IsNullOrEmpty(options.Body))
+                {
+                    string contentType = "text/plain; charset=utf-8";
+                    if (options.Headers != null && options.Headers.TryGetValue("content-type", out var ct))
+                    {
+                        contentType = ct;
+                    }
+                    var bodyString = options.Body;
+                    var mediaType = contentType.Split(';')[0].Trim();
+                    var charset = Encoding.UTF8;
+                    configBuilder.RequestBodyFactory(() =>
+                        new StringContent(bodyString, charset, mediaType));
+                }
+            }
+
+            _eventSource = new EventSource(configBuilder.Build());
+            _eventSource.MessageReceived += OnMessageReceived;
+            _eventSource.Error += OnError;
+            // Deliberately not subscribing to CommentReceived: the "comments" capability is
+            // not declared because .NET EventSource returns comment strings with the leading
+            // colon still attached, which does not match the harness's expected shape. If we
+            // forwarded comments anyway, the harness would receive unexpected comment messages
+            // in tests that assume no comment reporting.
+
+            // Fire-and-forget: EventSource fires events on its own internal thread as data arrives.
+            _ = _eventSource.StartAsync();
+        }
+
+        public bool DoCommand(string command)
+        {
+            _logger.Info("Test harness sent command: {0}", command);
+            if (command == "restart")
+            {
+                _eventSource.Restart(false);
+                return true;
+            }
+            return false;
+        }
+
+        public void Close()
+        {
+            _closed = true;
+            _eventSource.MessageReceived -= OnMessageReceived;
+            _eventSource.Error -= OnError;
+            _eventSource.Close();
+            _callbackClient.Dispose();
+            _logger.Info("Test ended");
+        }
+
+        private void OnMessageReceived(object sender, MessageReceivedEventArgs e)
+        {
+            _logger.Info("Received event from stream ({0})", e.EventName);
+            var msg = new Message
+            {
+                Kind = "event",
+                Event = new EventMessage
+                {
+                    Type = e.EventName,
+                    Data = e.Message.Data,
+                    Id = e.Message.LastEventId,
+                },
+            };
+            SendCallback(msg);
+        }
+
+        private void OnError(object sender, ExceptionEventArgs e)
+        {
+            _logger.Info("Received error from stream: {0}", e.Exception);
+            SendCallback(new Message { Kind = "error", Error = e.Exception.ToString() });
+        }
+
+        private void SendCallback(Message message)
+        {
+            if (_closed)
+            {
+                return;
+            }
+            var counter = Interlocked.Increment(ref _callbackMessageCounter);
+            var url = _options.CallbackUrl + "/" + counter;
+            var json = JsonSerializer.Serialize(message, JsonOptions);
+            try
+            {
+                using var content = new StringContent(json, Encoding.UTF8);
+                content.Headers.ContentType = new MediaTypeHeaderValue("application/json");
+                var resp = _callbackClient.PostAsync(url, content).GetAwaiter().GetResult();
+                if ((int)resp.StatusCode >= 300)
+                {
+                    _logger.Error("Callback to {0} returned HTTP {1}", url, (int)resp.StatusCode);
+                }
+            }
+            catch (Exception ex)
+            {
+                _logger.Error("Callback to {0} failed: {1}", url, ex.GetType().Name);
+            }
+        }
+    }
+}

contract-tests/TestService.cs

@@ -0,0 +1,138 @@
+using System.Collections.Concurrent;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using LaunchDarkly.Logging;
+using LaunchDarkly.TestHelpers.HttpTest;
+
+namespace TestService
+{
+    /// <summary>
+    /// HTTP entry point for the SSE contract-tests service. Implements the endpoints described in
+    /// launchdarkly/sse-contract-tests/docs/service_spec.md for the LaunchDarkly.EventSource
+    /// library. This is the .NET analogue of ssetest.TestService in okhttp-eventsource's
+    /// contract-tests service.
+    /// </summary>
+    public class Program
+    {
+        const int Port = 8000;
+
+        public static void Main(string[] args)
+        {
+            var quitSignal = new EventWaitHandle(false, EventResetMode.AutoReset);
+
+            var app = new Webapp(quitSignal);
+            var server = HttpServer.Start(Port, app.Handler);
+            server.Recorder.Enabled = false;
+
+            System.Console.WriteLine("Listening on port {0}", Port);
+
+            quitSignal.WaitOne();
+            server.Dispose();
+        }
+    }
+
+    public class Webapp
+    {
+        // Capabilities the LaunchDarkly.EventSource library implements correctly against the
+        // sse-contract-tests harness. Each entry here was verified end-to-end by running the
+        // harness's tests for that capability. Deliberately omitted:
+        //
+        //   * "comments" -- CommentReceived fires with the raw line including the leading colon;
+        //     the harness expects the colon stripped. See EventParser.cs where colonPos==0 stores
+        //     `line` (with colon) as ValueString. Real .NET SDK behavior; not fixed here.
+        //   * "event-type-listeners" -- not applicable; MessageReceived fires for all event types
+        //     without explicit registration.
+        //   * "server-directed-shutdown-request" -- .NET retries on 204 instead of halting.
+        //     EventSourceService.ValidateResponse throws on 204, EventSource.cs top-level loop
+        //     re-enters on Closed (not Shutdown) state. Real .NET SDK behavior; not fixed here.
+        private static readonly string[] Capabilities = new[]
+        {
+            "bom",
+            "headers",
+            "last-event-id",
+            "payload-size-stress-testable",
+            "post",
+            "read-timeout",
+            "report",
+            "restart",
+        };
+
+        public readonly Handler Handler;
+
+        private readonly ILogAdapter _logging = Logs.ToConsole;
+        private readonly ConcurrentDictionary<string, StreamEntity> _streams =
+            new ConcurrentDictionary<string, StreamEntity>();
+        private readonly EventWaitHandle _quitSignal;
+        private volatile int _lastStreamId = 0;
+
+        public Webapp(EventWaitHandle quitSignal)
+        {
+            _quitSignal = quitSignal;
+
+            var service = new SimpleJsonService();
+            Handler = service.Handler;
+
+            service.Route(HttpMethod.Get, "/", GetStatus);
+            service.Route(HttpMethod.Delete, "/", ForceQuit);
+            service.Route<StreamOptions>(HttpMethod.Post, "/", PostCreateStream);
+            service.Route<CommandParams>(HttpMethod.Post, "/streams/(.*)", PostStreamCommand);
+            service.Route(HttpMethod.Delete, "/streams/(.*)", DeleteStream);
+        }
+
+        SimpleResponse<Status> GetStatus(IRequestContext context) =>
+            SimpleResponse.Of(200, new Status { Capabilities = Capabilities });
+
+        SimpleResponse ForceQuit(IRequestContext context)
+        {
+            _logging.Logger("").Info("Test harness has told us to exit");
+
+            // The web server won't send the response till we return, so we'll defer the actual shutdown
+            _ = Task.Run(async () =>
+            {
+                await Task.Delay(100);
+                _quitSignal.Set();
+            });
+
+            return SimpleResponse.Of(204);
+        }
+
+        SimpleResponse PostCreateStream(IRequestContext context, StreamOptions opts)
+        {
+            var stream = new StreamEntity(opts, _logging);
+
+            var id = Interlocked.Increment(ref _lastStreamId);
+            var streamId = id.ToString();
+            _streams[streamId] = stream;
+
+            var resourceUrl = "/streams/" + streamId;
+            return SimpleResponse.Of(201).WithHeader("Location", resourceUrl);
+        }
+
+        SimpleResponse PostStreamCommand(IRequestContext context, CommandParams cmd)
+        {
+            var id = context.GetPathParam(0);
+            if (!_streams.TryGetValue(id, out var stream))
+            {
+                return SimpleResponse.Of(404);
+            }
+            if (!stream.DoCommand(cmd.Command))
+            {
+                return SimpleResponse.Of(400);
+            }
+            return SimpleResponse.Of(204);
+        }
+
+        SimpleResponse DeleteStream(IRequestContext context)
+        {
+            var id = context.GetPathParam(0);
+            if (!_streams.TryGetValue(id, out var stream))
+            {
+                return SimpleResponse.Of(404);
+            }
+            stream.Close();
+            _streams.TryRemove(id, out _);
+            return SimpleResponse.Of(204);
+        }
+    }
+}