增加 sse服务端测试

test/OrderSupportTicket.TestServer/Services/TicketEventHub.cs

@@ -0,0 +1,92 @@
+using System.Collections.Concurrent;
+using System.Threading.Channels;
+using OrderSupportTicket.TestServer.Contracts;
+
+namespace OrderSupportTicket.TestServer.Services;
+
+/// <summary>
+/// 纯内存 SSE 订阅中心。进程退出后事件和订阅全部丢失。
+/// </summary>
+public sealed class TicketEventHub
+{
+    private readonly ConcurrentDictionary<Guid, Subscription> _subscriptions = new();
+
+    public int SubscriberCount => _subscriptions.Count;
+
+    public TicketEventSubscription Subscribe(
+        string purchaseAccount,
+        IReadOnlyCollection<string> appIds)
+    {
+        var id = Guid.NewGuid();
+        var channel = Channel.CreateUnbounded<TicketRealtimeNotificationDto>(
+            new UnboundedChannelOptions
+            {
+                SingleReader = true,
+                SingleWriter = false,
+                AllowSynchronousContinuations = false
+            });
+        var subscription = new Subscription(
+            purchaseAccount,
+            new HashSet<string>(appIds, StringComparer.OrdinalIgnoreCase),
+            channel);
+        _subscriptions[id] = subscription;
+        return new TicketEventSubscription(
+            channel.Reader,
+            () => _subscriptions.TryRemove(id, out _));
+    }
+
+    public void Publish(
+        string purchaseAccount,
+        TicketRealtimeNotificationDto notification)
+    {
+        foreach (var subscription in _subscriptions.Values)
+        {
+            if (!string.Equals(
+                    subscription.PurchaseAccount,
+                    purchaseAccount,
+                    StringComparison.Ordinal))
+            {
+                continue;
+            }
+
+            if (!subscription.AppIds.Contains(notification.AppId))
+            {
+                continue;
+            }
+
+            subscription.Channel.Writer.TryWrite(notification);
+        }
+    }
+
+    private sealed record Subscription(
+        string PurchaseAccount,
+        HashSet<string> AppIds,
+        Channel<TicketRealtimeNotificationDto> Channel);
+}
+
+public sealed class TicketEventSubscription : IDisposable
+{
+    private readonly Action _unsubscribe;
+    private int _disposed;
+
+    internal TicketEventSubscription(
+        ChannelReader<TicketRealtimeNotificationDto> reader,
+        Action unsubscribe)
+    {
+        Reader = reader;
+        _unsubscribe = unsubscribe;
+    }
+
+    public ChannelReader<TicketRealtimeNotificationDto> Reader { get; }
+
+    public void Dispose()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0)
+        {
+            return;
+        }
+
+        _unsubscribe();
+    }
+}
+