feat(server): member last seen and a live per-account space activity stream
This commit is contained in:
1 parent
78b4e2aa63
commit
5043c798ff
3 files changed
+66
-2
No files matched your search
@@ -159,6 +159,9 @@ public sealed class MemberBody
|
||||
/// <summary>该账号在本频道登记过的设备(终端)数;设备只是账号的属性,不是独立成员。</summary>
|
||||
[JsonPropertyName("devices")] public required int Devices { get; init; }
|
||||
[JsonPropertyName("online_devices")] public required int OnlineDevices { get; init; }
|
||||
|
||||
/// <summary>该账号任一设备最近一次活跃的 Unix 时间戳(登录时是秒,活跃续期后是毫秒,客户端按量级区分);没有登录过设备时为 0。</summary>
|
||||
[JsonPropertyName("last_seen")] public required long LastSeen { get; init; }
|
||||
}
|
||||
|
||||
public sealed class AccountProfileRequest
|
||||
|
||||
@@ -12,6 +12,10 @@ public sealed class ChannelHub : IChannelEvents
|
||||
private readonly object _gate = new();
|
||||
private readonly Dictionary<Id16, Signal> _signals = [];
|
||||
|
||||
// 账号级的「有频道前进了」信号:不管有没有人订阅这个频道都会记录,供空间列表上的未读提示用
|
||||
private readonly Dictionary<Id16, long> _advanced = [];
|
||||
private TaskCompletionSource _anyAdvanced = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
|
||||
// 在线状态:频道 -> 账号 -> 账号设备 -> 活动订阅数。只记账号。
|
||||
private readonly Dictionary<Id16, Dictionary<string, Dictionary<string, int>>> _presence = [];
|
||||
|
||||
@@ -66,7 +70,24 @@ public sealed class ChannelHub : IChannelEvents
|
||||
public void Dispose() => Interlocked.Exchange(ref _release, null)?.Invoke();
|
||||
}
|
||||
|
||||
public void Advanced(Id16 channelId, long seq) => Publish(channelId, s => s.Seq = Math.Max(s.Seq, seq));
|
||||
public void Advanced(Id16 channelId, long seq)
|
||||
{
|
||||
TaskCompletionSource old;
|
||||
lock (_gate)
|
||||
{
|
||||
_advanced[channelId] = Math.Max(_advanced.GetValueOrDefault(channelId), seq);
|
||||
old = _anyAdvanced;
|
||||
_anyAdvanced = new(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
}
|
||||
old.TrySetResult();
|
||||
Publish(channelId, s => s.Seq = Math.Max(s.Seq, seq));
|
||||
}
|
||||
|
||||
/// <summary>先取这个任务、再读 <see cref="AdvancedSeq"/>,最后等待任务:不会漏掉中间的前进。</summary>
|
||||
public Task AnyAdvanced { get { lock (_gate) return _anyAdvanced.Task; } }
|
||||
|
||||
/// <summary>本进程启动以来该频道见到的最大序号;没有见到过为 0。</summary>
|
||||
public long AdvancedSeq(Id16 channelId) { lock (_gate) return _advanced.GetValueOrDefault(channelId); }
|
||||
public void SettingsChanged(Id16 channelId) => Publish(channelId, s => s.SettingsVersion++);
|
||||
public void MembersChanged(Id16 channelId) => Publish(channelId, s => s.MembersVersion++);
|
||||
public void ChannelDeleted(Id16 channelId) => Publish(channelId, s => s.Deleted = true);
|
||||
|
||||
@@ -12,6 +12,44 @@ public static partial class Endpoints
|
||||
if (!TryAccount(ctx, accounts, out var username)) throw new WebDropException(ErrorCode.AccountRequired);
|
||||
return Json(svc.AccountSpaces(username).ToList(), ApiJson.Default.ListChannelInfo);
|
||||
});
|
||||
// 空间列表的实时提示:账号所在空间有新内容就推一条 {id,last_seq}。只含序号,不含内容
|
||||
api.MapGet("/accounts/spaces/events", async (HttpContext ctx, ChannelService svc, ChannelHub hub) =>
|
||||
{
|
||||
if (!TryAccount(ctx, accounts, out var username)) throw new WebDropException(ErrorCode.AccountRequired);
|
||||
var ct = ctx.RequestAborted;
|
||||
ctx.Response.ContentType = "text/event-stream";
|
||||
ctx.Response.Headers["X-Accel-Buffering"] = "no";
|
||||
await ctx.Response.StartAsync(ct);
|
||||
var sent = new Dictionary<Id16, long>();
|
||||
try
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
// 每次唤醒都重新核对登录:退出登录、被移除设备后,这条流立刻结束
|
||||
if (!TryAccount(ctx, accounts, out var current) || current != username)
|
||||
{
|
||||
await ctx.Response.WriteAsync("event: error\ndata: {\"error\":\"account_required\"}\n\n", ct);
|
||||
break;
|
||||
}
|
||||
var changed = hub.AnyAdvanced;
|
||||
foreach (var id in svc.AccountSpaceIds(username))
|
||||
{
|
||||
var seq = hub.AdvancedSeq(id);
|
||||
// 第一次只记下当前值:连接建立前的进展客户端已经从列表里拿到了
|
||||
if (!sent.TryGetValue(id, out var known)) { sent[id] = seq; continue; }
|
||||
if (seq <= known) continue;
|
||||
sent[id] = seq;
|
||||
await ctx.Response.WriteAsync($"event: space\ndata: {{\"id\":\"{id.ToBase64Url()}\",\"last_seq\":{seq}}}\n\n", ct);
|
||||
}
|
||||
await ctx.Response.Body.FlushAsync(ct);
|
||||
var idle = Task.Delay(TimeSpan.FromSeconds(20), ct);
|
||||
// 久无变化就发心跳,客户端据此判断连接还活着
|
||||
if (await Task.WhenAny(changed, idle) == idle) { await ctx.Response.WriteAsync("event: ping\ndata: 1\n\n", ct); await ctx.Response.Body.FlushAsync(ct); }
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) { }
|
||||
return Results.Empty;
|
||||
});
|
||||
api.MapGet("/accounts/channels", (HttpContext ctx, ChannelService svc) =>
|
||||
{
|
||||
if (!TryAccount(ctx, accounts, out var username)) throw new WebDropException(ErrorCode.AccountRequired);
|
||||
@@ -54,8 +92,10 @@ public static partial class Endpoints
|
||||
{
|
||||
var (online, onlineDevices) = hub.Presence(channelId, member.Username);
|
||||
// 设备是账号的属性:总数取账号的登录设备,在线数取该账号在本频道有订阅的设备
|
||||
var devices = accounts.Devices(member.Username, null);
|
||||
return new MemberBody { Username = member.Username, Role = member.Role, JoinedAt = member.JoinedAt,
|
||||
Online = online, Devices = Math.Max(accounts.Devices(member.Username, null).Count, onlineDevices), OnlineDevices = onlineDevices };
|
||||
Online = online, Devices = Math.Max(devices.Count, onlineDevices), OnlineDevices = onlineDevices,
|
||||
LastSeen = devices.Count == 0 ? 0 : devices.Max(device => device.LastSeen) };
|
||||
}).ToList();
|
||||
return Json(members, ApiJson.Default.ListMemberBody);
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user