[codex] Improve high-event hot paths

benchmarks/NetDaemon.PerformanceBenchmarks/Benchmarks/QueuedObservableBenchmarks.cs

@@ -0,0 +1,38 @@
+using System.Reactive.Subjects;
+using BenchmarkDotNet.Attributes;
+using Microsoft.Extensions.Logging.Abstractions;
+using NetDaemon.HassModel.Internal;
+
+namespace NetDaemon.PerformanceBenchmarks.Benchmarks;
+
+[MemoryDiagnoser]
+public class QueuedObservableBenchmarks
+{
+    private const int EventCount = 10_000;
+
+    [Params(1, 10, 50)]
+    public int AppScopes { get; set; }
+
+    [Benchmark]
+    public async Task FanOutToAppScopedQueues()
+    {
+        using var source = new Subject<int>();
+        var queues = new QueuedObservable<int>[AppScopes];
+
+        for (var i = 0; i < queues.Length; i++)
+        {
+            queues[i] = new QueuedObservable<int>(source, NullLogger.Instance);
+            queues[i].Subscribe(static _ => { });
+        }
+
+        for (var i = 0; i < EventCount; i++)
+        {
+            source.OnNext(i);
+        }
+
+        foreach (var queue in queues)
+        {
+            await queue.DisposeAsync().ConfigureAwait(false);
+        }
+    }
+}

benchmarks/NetDaemon.PerformanceBenchmarks/Benchmarks/ResultDispatchBenchmarks.cs

@@ -0,0 +1,57 @@
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Reactive.Threading.Tasks;
+using System.Text.Json;
+using BenchmarkDotNet.Attributes;
+using NetDaemon.Client.Internal.HomeAssistant.Commands;
+using NetDaemon.PerformanceBenchmarks.Support;
+
+namespace NetDaemon.PerformanceBenchmarks.Benchmarks;
+
+[MemoryDiagnoser]
+public class ResultDispatchBenchmarks
+{
+    private HassMessage[] _results = [];
+
+    [Params(1, 10, 100, 1000)]
+    public int PendingCommands { get; set; }
+
+    [GlobalSetup]
+    public void Setup()
+    {
+        _results = Enumerable.Range(1, PendingCommands)
+            .Select(id => JsonSerializer.Deserialize<HassMessage>(HassPayloadFactory.ResultMessage(id))!)
+            .ToArray();
+    }
+
+    [Benchmark(Baseline = true)]
+    public async Task PerCommandRxFilterSubscriptions()
+    {
+        using var subject = new Subject<HassMessage>();
+        var tasks = Enumerable.Range(1, PendingCommands)
+            .Select(id => subject.Where(n => n.Type == "result" && n.Id == id).FirstAsync().ToTask())
+            .ToArray();
+
+        foreach (var result in _results)
+        {
+            subject.OnNext(result);
+        }
+
+        await Task.WhenAll(tasks).ConfigureAwait(false);
+    }
+
+    [Benchmark]
+    public void DictionaryResultDispatch()
+    {
+        var pending = Enumerable.Range(1, PendingCommands)
+            .ToDictionary(static id => id, static _ => new TaskCompletionSource<HassMessage>(TaskCreationOptions.RunContinuationsAsynchronously));
+
+        foreach (var result in _results)
+        {
+            if (pending.Remove(result.Id, out var completionSource))
+            {
+                completionSource.SetResult(result);
+            }
+        }
+    }
+}

benchmarks/NetDaemon.PerformanceBenchmarks/Benchmarks/StateChangeJsonBenchmarks.cs

@@ -0,0 +1,45 @@
+using System.Text.Json;
+using BenchmarkDotNet.Attributes;
+using NetDaemon.Client.HomeAssistant.Model;
+using NetDaemon.PerformanceBenchmarks.Support;
+
+namespace NetDaemon.PerformanceBenchmarks.Benchmarks;
+
+[MemoryDiagnoser]
+public class StateChangeJsonBenchmarks : IDisposable
+{
+    private readonly JsonDocument _document;
+    private readonly JsonElement _eventData;
+
+    public StateChangeJsonBenchmarks()
+    {
+        _document = JsonDocument.Parse(HassPayloadFactory.StateChangedEvent(1));
+        _eventData = _document.RootElement.GetProperty("event").GetProperty("data");
+    }
+
+    [Benchmark(Baseline = true)]
+    public string? ExtractEntityIdAndLazyNewState()
+    {
+        var entityId = _eventData.GetProperty("entity_id").GetString();
+        _ = _eventData.GetProperty("new_state");
+        return entityId;
+    }
+
+    [Benchmark]
+    public HassState? ForceDeserializeNewState()
+    {
+        return _eventData.GetProperty("new_state").Deserialize<HassState>();
+    }
+
+    public void Dispose()
+    {
+        Dispose(true);
+        GC.SuppressFinalize(this);
+    }
+
+    protected virtual void Dispose(bool disposing)
+    {
+        if (disposing)
+            _document.Dispose();
+    }
+}

benchmarks/NetDaemon.PerformanceBenchmarks/Benchmarks/WebSocketPipelineBenchmarks.cs

@@ -0,0 +1,44 @@
+using BenchmarkDotNet.Attributes;
+using NetDaemon.Client.Internal.HomeAssistant.Commands;
+using NetDaemon.Client.Internal.Net;
+using NetDaemon.PerformanceBenchmarks.Support;
+
+namespace NetDaemon.PerformanceBenchmarks.Benchmarks;
+
+[MemoryDiagnoser]
+public class WebSocketPipelineBenchmarks
+{
+    private const int CoalescedEventCount = 64;
+    private readonly string _singleEvent = HassPayloadFactory.StateChangedEvent(1);
+    private readonly string _coalescedEvents = HassPayloadFactory.CoalescedStateChangedEvents(CoalescedEventCount);
+    private readonly object _serviceCommand = new { id = 42, type = "call_service", domain = "light", service = "turn_on" };
+
+    [Benchmark(Baseline = true)]
+    public async Task<HassMessage[]> ReadSingleEvent()
+    {
+        var websocket = new BenchmarkWebSocketClient();
+        websocket.EnqueueJson(_singleEvent);
+        var pipeline = new WebSocketClientTransportPipeline(websocket);
+
+        return await pipeline.GetNextMessagesAsync<HassMessage>(CancellationToken.None).ConfigureAwait(false);
+    }
+
+    [Benchmark]
+    public async Task<HassMessage[]> ReadCoalescedEvents()
+    {
+        var websocket = new BenchmarkWebSocketClient();
+        websocket.EnqueueJson(_coalescedEvents);
+        var pipeline = new WebSocketClientTransportPipeline(websocket);
+
+        return await pipeline.GetNextMessagesAsync<HassMessage>(CancellationToken.None).ConfigureAwait(false);
+    }
+
+    [Benchmark]
+    public async Task SendServiceCommand()
+    {
+        var websocket = new BenchmarkWebSocketClient();
+        var pipeline = new WebSocketClientTransportPipeline(websocket);
+
+        await pipeline.SendMessageAsync(_serviceCommand, CancellationToken.None).ConfigureAwait(false);
+    }
+}

benchmarks/NetDaemon.PerformanceBenchmarks/NetDaemon.PerformanceBenchmarks.csproj

@@ -0,0 +1,21 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <OutputType>Exe</OutputType>
+    <IsPackable>false</IsPackable>
+    <Nullable>enable</Nullable>
+    <ImplicitUsings>enable</ImplicitUsings>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="BenchmarkDotNet" Version="0.15.8" />
+    <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.7" />
+    <PackageReference Include="System.Reactive" Version="6.1.0" />
+  </ItemGroup>
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\src\Client\NetDaemon.HassClient\NetDaemon.Client.csproj" />
+    <ProjectReference Include="..\..\src\HassModel\NetDaemon.HassModel\NetDaemon.HassModel.csproj" />
+  </ItemGroup>
+
+</Project>

benchmarks/NetDaemon.PerformanceBenchmarks/Program.cs

@@ -0,0 +1,7 @@
+using BenchmarkDotNet.Configs;
+using BenchmarkDotNet.Running;
+
+var artifactsPath = Path.Combine(AppContext.BaseDirectory, "BenchmarkDotNet.Artifacts");
+var config = DefaultConfig.Instance.WithArtifactsPath(artifactsPath);
+
+BenchmarkSwitcher.FromAssembly(typeof(Program).Assembly).Run(args, config);

benchmarks/NetDaemon.PerformanceBenchmarks/Support/BenchmarkWebSocketClient.cs

@@ -0,0 +1,77 @@
+using System.Net.WebSockets;
+using System.Text;
+using NetDaemon.Client.Internal.Net;
+
+namespace NetDaemon.PerformanceBenchmarks.Support;
+
+internal sealed class BenchmarkWebSocketClient : IWebSocketClient
+{
+    private readonly Queue<byte[]> _responses = new();
+    private byte[]? _currentResponse;
+    private int _currentResponseOffset;
+
+    public WebSocketState State { get; private set; } = WebSocketState.Open;
+
+    public WebSocketCloseStatus? CloseStatus { get; private set; }
+
+    public ValueTask DisposeAsync()
+    {
+        State = WebSocketState.Closed;
+        return ValueTask.CompletedTask;
+    }
+
+    public Task ConnectAsync(Uri uri, CancellationToken cancellationToken) => Task.CompletedTask;
+
+    public Task CloseAsync(WebSocketCloseStatus closeStatus, string statusDescription, CancellationToken cancellationToken)
+    {
+        CloseStatus = closeStatus;
+        State = WebSocketState.Closed;
+        return Task.CompletedTask;
+    }
+
+    public Task CloseOutputAsync(WebSocketCloseStatus closeStatus, string statusDescription, CancellationToken cancellationToken)
+    {
+        CloseStatus = closeStatus;
+        State = WebSocketState.Closed;
+        return Task.CompletedTask;
+    }
+
+    public Task SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken)
+    {
+        SentBytes += buffer.Count;
+        SentMessages++;
+        return Task.CompletedTask;
+    }
+
+    public ValueTask SendAsync(ReadOnlyMemory<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken)
+    {
+        SentBytes += buffer.Length;
+        SentMessages++;
+        return ValueTask.CompletedTask;
+    }
+
+    public ValueTask<ValueWebSocketReceiveResult> ReceiveAsync(Memory<byte> buffer, CancellationToken cancellationToken)
+    {
+        _currentResponse ??= _responses.Dequeue();
+
+        var remaining = _currentResponse.Length - _currentResponseOffset;
+        var count = Math.Min(remaining, buffer.Length);
+        _currentResponse.AsMemory(_currentResponseOffset, count).CopyTo(buffer);
+        _currentResponseOffset += count;
+
+        var endOfMessage = _currentResponseOffset == _currentResponse.Length;
+        if (endOfMessage)
+        {
+            _currentResponse = null;
+            _currentResponseOffset = 0;
+        }
+
+        return ValueTask.FromResult(new ValueWebSocketReceiveResult(count, WebSocketMessageType.Text, endOfMessage));
+    }
+
+    public int SentMessages { get; private set; }
+
+    public int SentBytes { get; private set; }
+
+    public void EnqueueJson(string json) => _responses.Enqueue(Encoding.UTF8.GetBytes(json));
+}

benchmarks/NetDaemon.PerformanceBenchmarks/Support/HassPayloadFactory.cs

@@ -0,0 +1,64 @@
+using System.Text;
+
+namespace NetDaemon.PerformanceBenchmarks.Support;
+
+internal static class HassPayloadFactory
+{
+    public static string StateChangedEvent(int index)
+    {
+        return $$"""
+            {
+              "id": 2,
+              "type": "event",
+              "event": {
+                "event_type": "state_changed",
+                "data": {
+                  "entity_id": "sensor.perf_{{index}}",
+                  "old_state": {
+                    "entity_id": "sensor.perf_{{index}}",
+                    "state": "{{index - 1}}",
+                    "attributes": { "unit_of_measurement": "W", "friendly_name": "Perf {{index}}" },
+                    "last_changed": "2026-06-18T10:00:00Z",
+                    "last_updated": "2026-06-18T10:00:00Z"
+                  },
+                  "new_state": {
+                    "entity_id": "sensor.perf_{{index}}",
+                    "state": "{{index}}",
+                    "attributes": { "unit_of_measurement": "W", "friendly_name": "Perf {{index}}" },
+                    "last_changed": "2026-06-18T10:00:01Z",
+                    "last_updated": "2026-06-18T10:00:01Z",
+                    "context": { "id": "ctx-{{index}}", "parent_id": null, "user_id": null }
+                  }
+                },
+                "origin": "LOCAL",
+                "time_fired": "2026-06-18T10:00:01Z"
+              }
+            }
+            """;
+    }
+
+    public static string CoalescedStateChangedEvents(int count)
+    {
+        var builder = new StringBuilder(count * 700);
+        builder.Append('[');
+        for (var i = 0; i < count; i++)
+        {
+            if (i > 0) builder.Append(',');
+            builder.Append(StateChangedEvent(i));
+        }
+        builder.Append(']');
+        return builder.ToString();
+    }
+
+    public static string ResultMessage(int id)
+    {
+        return $$"""
+            {
+              "id": {{id}},
+              "type": "result",
+              "success": true,
+              "result": { "ok": true }
+            }
+            """;
+    }
+}