71 lines
3.1 KiB
C#
71 lines
3.1 KiB
C#
using System.Collections.Concurrent;
|
|
using System.Text.Json.Serialization;
|
|
using System.Threading.Channels;
|
|
using WxAgent.Core;
|
|
|
|
namespace WxAgent.Service;
|
|
|
|
public sealed record AgentEvent(
|
|
string EventId,
|
|
string AccountId,
|
|
string? Session,
|
|
string Kind,
|
|
string Summary,
|
|
DateTimeOffset At,
|
|
[property: JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] string? ChatId = null,
|
|
[property: JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] ReportingChatType? ChatType = null,
|
|
[property: JsonIgnore(Condition = JsonIgnoreCondition.WhenWritingNull)] string? Content = null);
|
|
|
|
public sealed class EventHub
|
|
{
|
|
private readonly ConcurrentDictionary<string, Subscription> subscriptions = new();
|
|
private readonly object gate = new();
|
|
private readonly Queue<HistoryEntry> history = new();
|
|
private const int MaxHistory = 200;
|
|
|
|
public (string SubscriptionId, IReadOnlyList<AgentEvent> Replay, bool Gap) Subscribe(string principal, string? after)
|
|
{
|
|
lock (gate)
|
|
{
|
|
var scopedHistory = history.Where(entry => entry.Principal == principal).Select(entry => entry.Event).ToArray();
|
|
var replay = after is null ? [] : scopedHistory.SkipWhile(e => e.EventId != after).Skip(1).ToArray();
|
|
var gap = after is not null && !scopedHistory.Any(e => e.EventId == after);
|
|
var id = Convert.ToHexString(System.Security.Cryptography.RandomNumberGenerator.GetBytes(16));
|
|
subscriptions[id] = new Subscription(principal);
|
|
return (id, replay, gap);
|
|
}
|
|
}
|
|
|
|
public void Publish(string principal, AgentEvent value)
|
|
{
|
|
lock (gate)
|
|
{
|
|
history.Enqueue(new HistoryEntry(principal, value));
|
|
while (history.Count > MaxHistory) history.Dequeue();
|
|
foreach (var subscription in subscriptions.Values.Where(s => s.Principal == principal))
|
|
{
|
|
if (subscription.Channel.Writer.TryWrite(value)) continue;
|
|
// ponytail: one bounded channel per subscriber; if it overflows, emit an explicit resync marker.
|
|
subscription.Channel.Reader.TryRead(out _);
|
|
subscription.Channel.Writer.TryWrite(new AgentEvent(
|
|
Convert.ToHexString(System.Security.Cryptography.RandomNumberGenerator.GetBytes(16)),
|
|
value.AccountId, value.Session, "gap", "resync required", DateTimeOffset.UtcNow));
|
|
}
|
|
}
|
|
}
|
|
|
|
public bool TryGet(string id, string principal, out Subscription subscription) =>
|
|
subscriptions.TryGetValue(id, out subscription!) && subscription.Principal == principal;
|
|
|
|
public void Remove(string id) => subscriptions.TryRemove(id, out _);
|
|
|
|
private sealed record HistoryEntry(string Principal, AgentEvent Event);
|
|
|
|
public sealed class Subscription(string principal)
|
|
{
|
|
public string Principal { get; } = principal;
|
|
public Channel<AgentEvent> Channel { get; } = System.Threading.Channels.Channel.CreateBounded<AgentEvent>(new BoundedChannelOptions(100)
|
|
{ FullMode = BoundedChannelFullMode.Wait, SingleReader = true, SingleWriter = false });
|
|
}
|
|
}
|