Add ActiveDevice + DeviceRegistry + HubEnvelopes + IDeviceClient/Channel implementations
This commit is contained in:
@@ -0,0 +1,25 @@
|
|||||||
|
using System.Net.WebSockets;
|
||||||
|
using backend.DeviceHub;
|
||||||
|
using FluentAssertions;
|
||||||
|
using Xunit;
|
||||||
|
|
||||||
|
namespace backend.tests;
|
||||||
|
|
||||||
|
public class DeviceRegistryTests
|
||||||
|
{
|
||||||
|
[Fact]
|
||||||
|
public void Register_unregister_round_trip()
|
||||||
|
{
|
||||||
|
var reg = new DeviceRegistry();
|
||||||
|
var id = Guid.NewGuid();
|
||||||
|
var dev = new ActiveDevice(id, new ClientWebSocket());
|
||||||
|
reg.TryRegister(dev).Should().BeTrue();
|
||||||
|
reg.TryRegister(dev).Should().BeFalse();
|
||||||
|
reg.IsOnline(id).Should().BeTrue();
|
||||||
|
reg.OnlineCount.Should().Be(1);
|
||||||
|
|
||||||
|
reg.Unregister(id).Should().BeTrue();
|
||||||
|
reg.IsOnline(id).Should().BeFalse();
|
||||||
|
reg.OnlineCount.Should().Be(0);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,14 +1,28 @@
|
|||||||
using System.Collections.Concurrent;
|
using System.Collections.Concurrent;
|
||||||
using System.Text.Json;
|
using System.Text.Json;
|
||||||
|
using backend.DeviceHub;
|
||||||
using backend.Tools;
|
using backend.Tools;
|
||||||
|
|
||||||
namespace backend.tests;
|
namespace backend.tests;
|
||||||
|
|
||||||
// Slim version for Tasks 9-15. Task 16 reopens this file to add IDeviceClient + envelope sending.
|
public class FakeDeviceChannel : IDeviceChannel, IDeviceClient
|
||||||
public class FakeDeviceChannel : IDeviceChannel
|
|
||||||
{
|
{
|
||||||
|
public List<HubEnvelope> SentEnvelopes { get; } = new();
|
||||||
|
public List<byte[]> SentBinary { get; } = new();
|
||||||
public ConcurrentQueue<JsonElement> PiToolReplies { get; } = new();
|
public ConcurrentQueue<JsonElement> PiToolReplies { get; } = new();
|
||||||
|
|
||||||
|
public Task SendEnvelopeAsync<T>(T envelope, CancellationToken ct) where T : HubEnvelope
|
||||||
|
{
|
||||||
|
SentEnvelopes.Add(envelope);
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task SendBinaryAsync(ReadOnlyMemory<byte> bytes, CancellationToken ct)
|
||||||
|
{
|
||||||
|
SentBinary.Add(bytes.ToArray());
|
||||||
|
return Task.CompletedTask;
|
||||||
|
}
|
||||||
|
|
||||||
public Task<JsonElement> CallPiToolAsync(string name, JsonElement args, CancellationToken ct)
|
public Task<JsonElement> CallPiToolAsync(string name, JsonElement args, CancellationToken ct)
|
||||||
{
|
{
|
||||||
if (PiToolReplies.TryDequeue(out var reply))
|
if (PiToolReplies.TryDequeue(out var reply))
|
||||||
|
|||||||
@@ -0,0 +1,112 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using System.Net.WebSockets;
|
||||||
|
using System.Text;
|
||||||
|
using System.Text.Json;
|
||||||
|
using System.Text.Json.Nodes;
|
||||||
|
using backend.Realtime;
|
||||||
|
using backend.Tools;
|
||||||
|
|
||||||
|
namespace backend.DeviceHub;
|
||||||
|
|
||||||
|
public interface IDeviceClient
|
||||||
|
{
|
||||||
|
Task SendEnvelopeAsync<T>(T envelope, CancellationToken ct) where T : HubEnvelope;
|
||||||
|
Task SendBinaryAsync(ReadOnlyMemory<byte> bytes, CancellationToken ct);
|
||||||
|
}
|
||||||
|
|
||||||
|
public class ActiveDevice : IDeviceChannel, IDeviceClient, ISessionOutput, IAsyncDisposable
|
||||||
|
{
|
||||||
|
public Guid DeviceId { get; }
|
||||||
|
public WebSocket Ws => _ws;
|
||||||
|
public RealtimeSession? CurrentSession { get; set; }
|
||||||
|
public CancellationTokenSource? SessionCts { get; set; }
|
||||||
|
|
||||||
|
private readonly WebSocket _ws;
|
||||||
|
private readonly SemaphoreSlim _sendLock = new(1, 1);
|
||||||
|
private readonly ConcurrentDictionary<string, TaskCompletionSource<JsonElement>> _pendingTools = new();
|
||||||
|
|
||||||
|
public ActiveDevice(Guid deviceId, WebSocket ws)
|
||||||
|
{
|
||||||
|
DeviceId = deviceId;
|
||||||
|
_ws = ws;
|
||||||
|
}
|
||||||
|
|
||||||
|
public async Task SendEnvelopeAsync<T>(T envelope, CancellationToken ct) where T : HubEnvelope
|
||||||
|
{
|
||||||
|
var json = JsonSerializer.Serialize(envelope, envelope.GetType());
|
||||||
|
await _sendLock.WaitAsync(ct);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _ws.SendAsync(Encoding.UTF8.GetBytes(json),
|
||||||
|
WebSocketMessageType.Text, endOfMessage: true, ct);
|
||||||
|
}
|
||||||
|
finally { _sendLock.Release(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
public async Task SendBinaryAsync(ReadOnlyMemory<byte> bytes, CancellationToken ct)
|
||||||
|
{
|
||||||
|
await _sendLock.WaitAsync(ct);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _ws.SendAsync(bytes, WebSocketMessageType.Binary, endOfMessage: true, ct);
|
||||||
|
}
|
||||||
|
finally { _sendLock.Release(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task WriteEnvelopeAsync(JsonObject envelope, CancellationToken ct)
|
||||||
|
{
|
||||||
|
var json = envelope.ToJsonString();
|
||||||
|
return WriteRawTextAsync(json, ct);
|
||||||
|
}
|
||||||
|
|
||||||
|
private async Task WriteRawTextAsync(string json, CancellationToken ct)
|
||||||
|
{
|
||||||
|
await _sendLock.WaitAsync(ct);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await _ws.SendAsync(Encoding.UTF8.GetBytes(json),
|
||||||
|
WebSocketMessageType.Text, endOfMessage: true, ct);
|
||||||
|
}
|
||||||
|
finally { _sendLock.Release(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
public Task WriteBinaryAsync(ReadOnlyMemory<byte> bytes, CancellationToken ct)
|
||||||
|
=> SendBinaryAsync(bytes, ct);
|
||||||
|
|
||||||
|
public Task<JsonElement> CallPiToolAsync(string name, JsonElement args, CancellationToken ct)
|
||||||
|
{
|
||||||
|
var callId = "srv_" + Guid.NewGuid().ToString("N")[..12];
|
||||||
|
var tcs = new TaskCompletionSource<JsonElement>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||||
|
_pendingTools[callId] = tcs;
|
||||||
|
|
||||||
|
var envelope = new ToolCallEnvelope("tool_call", callId, name, args);
|
||||||
|
_ = SendEnvelopeAsync(envelope, ct);
|
||||||
|
|
||||||
|
ct.Register(() => tcs.TrySetCanceled());
|
||||||
|
return tcs.Task;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void CompleteToolResult(ToolResultEnvelope env)
|
||||||
|
{
|
||||||
|
if (!_pendingTools.TryRemove(env.CallId, out var tcs)) return;
|
||||||
|
if (env.Ok && env.Result is JsonElement r) tcs.TrySetResult(r);
|
||||||
|
else
|
||||||
|
{
|
||||||
|
var msg = (env.Error ?? "unknown").Replace("\\", "\\\\").Replace("\"", "\\\"");
|
||||||
|
var doc = JsonDocument.Parse($$"""{"ok":false,"error":"{{msg}}"}""");
|
||||||
|
tcs.TrySetResult(doc.RootElement.Clone());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public async ValueTask DisposeAsync()
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
if (_ws.State == WebSocketState.Open)
|
||||||
|
await _ws.CloseAsync(WebSocketCloseStatus.NormalClosure, "bye", CancellationToken.None);
|
||||||
|
}
|
||||||
|
catch { }
|
||||||
|
_ws.Dispose();
|
||||||
|
_sendLock.Dispose();
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
|
||||||
|
namespace backend.DeviceHub;
|
||||||
|
|
||||||
|
public class DeviceRegistry
|
||||||
|
{
|
||||||
|
private readonly ConcurrentDictionary<Guid, ActiveDevice> _online = new();
|
||||||
|
|
||||||
|
public bool TryRegister(ActiveDevice dev)
|
||||||
|
=> _online.TryAdd(dev.DeviceId, dev);
|
||||||
|
|
||||||
|
public bool Unregister(Guid deviceId)
|
||||||
|
=> _online.TryRemove(deviceId, out _);
|
||||||
|
|
||||||
|
public ActiveDevice? Get(Guid deviceId)
|
||||||
|
=> _online.TryGetValue(deviceId, out var d) ? d : null;
|
||||||
|
|
||||||
|
public bool IsOnline(Guid deviceId) => _online.ContainsKey(deviceId);
|
||||||
|
|
||||||
|
public int OnlineCount => _online.Count;
|
||||||
|
}
|
||||||
@@ -0,0 +1,42 @@
|
|||||||
|
using System.Text.Json;
|
||||||
|
using System.Text.Json.Serialization;
|
||||||
|
|
||||||
|
namespace backend.DeviceHub;
|
||||||
|
|
||||||
|
public record HubEnvelope([property: JsonPropertyName("type")] string Type);
|
||||||
|
|
||||||
|
public record HelloEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("device_id")] Guid DeviceId,
|
||||||
|
[property: JsonPropertyName("client_version")] string ClientVersion) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record HelloAckEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("config")] JsonElement Config) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record PongEnvelope(string Type) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record WakeEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("at")] long? At) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record SessionStartedEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("conversation_id")] Guid ConversationId) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record SessionEndedEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("reason")] string Reason) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record AssistantDoneEnvelope(string Type) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record ToolCallEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("call_id")] string CallId,
|
||||||
|
[property: JsonPropertyName("name")] string Name,
|
||||||
|
[property: JsonPropertyName("arguments")] JsonElement Arguments) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record ToolResultEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("call_id")] string CallId,
|
||||||
|
[property: JsonPropertyName("ok")] bool Ok,
|
||||||
|
[property: JsonPropertyName("result")] JsonElement? Result,
|
||||||
|
[property: JsonPropertyName("error")] string? Error) : HubEnvelope(Type);
|
||||||
|
|
||||||
|
public record ErrorEnvelope(string Type,
|
||||||
|
[property: JsonPropertyName("code")] string Code,
|
||||||
|
[property: JsonPropertyName("message")] string Message,
|
||||||
|
[property: JsonPropertyName("fatal")] bool? Fatal) : HubEnvelope(Type);
|
||||||
@@ -40,6 +40,7 @@ builder.Services.AddSingleton<backend.Tools.ITool, backend.Tools.EndSessionTool>
|
|||||||
builder.Services.AddSingleton<backend.Tools.ITool, backend.Tools.SetVolumeTool>();
|
builder.Services.AddSingleton<backend.Tools.ITool, backend.Tools.SetVolumeTool>();
|
||||||
builder.Services.AddSingleton<backend.Tools.ToolRegistry>();
|
builder.Services.AddSingleton<backend.Tools.ToolRegistry>();
|
||||||
builder.Services.AddScoped<backend.Realtime.RealtimeSessionFactory>();
|
builder.Services.AddScoped<backend.Realtime.RealtimeSessionFactory>();
|
||||||
|
builder.Services.AddSingleton<backend.DeviceHub.DeviceRegistry>();
|
||||||
|
|
||||||
var app = builder.Build();
|
var app = builder.Build();
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user