385 lines
25 KiB
C#
385 lines
25 KiB
C#
using Evergreen.Health.Application.Health.Integrations.Ib43;
|
|
using Evergreen.Health.Application.Health.Entities;
|
|
using Evergreen.Health.Application.Health.Devices.Audio;
|
|
using Evergreen.Health.Application.Health.Devices.Climate;
|
|
using Evergreen.Health.Application.Health.Devices.Curtain;
|
|
using Evergreen.Health.Application.Health.Devices.Dehumidifier;
|
|
using Evergreen.Health.Application.Health.Devices.Environment;
|
|
using Evergreen.Health.Application.Health.Devices.Fragrance;
|
|
using Evergreen.Health.Application.Health.Integrations.HomeAssistant;
|
|
using Evergreen.Health.Application.Health.Integrations.MiAirX;
|
|
using Evergreen.Health.Application.Health.Scenes;
|
|
using Evergreen.Health.Application.Health.Sleep;
|
|
using SqlSugar;
|
|
using Microsoft.Extensions.Logging.Abstractions;
|
|
using Microsoft.Extensions.Options;
|
|
using MQTTnet;
|
|
using MQTTnet.Client;
|
|
using MQTTnet.Protocol;
|
|
using MQTTnet.Server;
|
|
using System.Net;
|
|
using System.Net.Sockets;
|
|
using System.Collections.Concurrent;
|
|
using System.Text.Json.Nodes;
|
|
using System.Text.Json;
|
|
using Xunit;
|
|
|
|
namespace Admin.NET.Test.Health;
|
|
|
|
public sealed class Ib43MqttIntegrationTests
|
|
{
|
|
private const string LeftBinding = "a98788f2aeeb40b68f145a182d99f832";
|
|
private const string TargetCurtain = "cover.test_curtain";
|
|
private const string TargetDehumidifier = "humidifier.test";
|
|
private const string HumiditySensor = "sensor.test_humidity";
|
|
private const string TargetClimate = "climate.test";
|
|
private const string TemperatureSensor = "sensor.test_temperature";
|
|
private const string FragranceButton = "button.test_fragrance";
|
|
private const string WakeFragranceButton = "button.test_wake_fragrance";
|
|
private const string Speaker = "media_player.test_speaker";
|
|
private const string TrackUrl = "https://example.test/sleep.mp3";
|
|
|
|
[Fact]
|
|
public async Task Real_MQTT_resubscribes_and_only_fresh_empty_to_occupied_triggers_after_broker_restart()
|
|
{
|
|
// 只监听 loopback，随机端口；不访问用户的 EMQX、数据库或家具。
|
|
using var reservation = new TcpListener(IPAddress.Loopback, 0);
|
|
reservation.Start();
|
|
var port = ((IPEndPoint)reservation.LocalEndpoint).Port;
|
|
reservation.Stop();
|
|
var factory = new MqttFactory();
|
|
using var broker = factory.CreateMqttServer(new MqttServerOptionsBuilder()
|
|
.WithDefaultEndpoint().WithDefaultEndpointPort(port)
|
|
.WithDefaultEndpointBoundIPAddress(IPAddress.Loopback).Build());
|
|
await broker.StartAsync();
|
|
var binding = new Ib43Binding { ProductKey = "evergreen-test", SerialNumber = "IB43SIM001" };
|
|
var options = Options.Create(new Ib43Options { Enabled = true, Host = "127.0.0.1", Port = port,
|
|
Username = "local-test", Password = "local-test-only", Devices = [binding] });
|
|
var store = new Ib43StateStore(options, new(), TimeProvider.System);
|
|
using var sink = new DatabaseSink(binding);
|
|
var gateway = new FakeHaGateway { CoverCloseResult = "closed" };
|
|
var openCurtain = new HomeAssistantState { EntityId = TargetCurtain, State = "open",
|
|
Attributes = JsonSerializer.SerializeToElement(new { supported_features = 2 }) };
|
|
var stoppedDehumidifier = new HomeAssistantState { EntityId = TargetDehumidifier, State = "off" };
|
|
gateway.States[TargetCurtain] = openCurtain;
|
|
gateway.States[TargetDehumidifier] = stoppedDehumidifier;
|
|
gateway.States[HumiditySensor] = new() { EntityId = HumiditySensor, State = "64", LastReported = DateTimeOffset.UtcNow,
|
|
Attributes = JsonSerializer.SerializeToElement(new { unit_of_measurement = "%" }) };
|
|
gateway.States[TemperatureSensor] = new() { EntityId = TemperatureSensor, State = "27", LastReported = DateTimeOffset.UtcNow,
|
|
Attributes = JsonSerializer.SerializeToElement(new { unit_of_measurement = "°C" }) };
|
|
gateway.States[FragranceButton] = new() { EntityId = FragranceButton, State = "unknown" };
|
|
gateway.States[WakeFragranceButton] = new() { EntityId = WakeFragranceButton, State = "unknown" };
|
|
gateway.States[Speaker] = new() { EntityId = Speaker, State = "idle" };
|
|
var haOptions = Options.Create(new HomeAssistantOptions
|
|
{ Entities = new() { Curtain = TargetCurtain, Dehumidifier = TargetDehumidifier, Humidity = HumiditySensor,
|
|
Climate = TargetClimate, Temperature = TemperatureSensor, FragranceQuickRelease = FragranceButton,
|
|
WakeFragranceQuickRelease = WakeFragranceButton, Speaker = Speaker } });
|
|
var execution = new DeviceExecutionOptions { VerificationDelay = TimeSpan.Zero };
|
|
var runner = new DeviceActionRunner(gateway, TimeProvider.System, execution);
|
|
var curtainAction = new BedPresenceCurtainCloseAction(new CurtainModule(runner, haOptions), runner, haOptions);
|
|
var preview = new BedPresenceDehumidifierPreview(gateway, haOptions, TimeProvider.System);
|
|
var dehumidifier = new BedPresenceDehumidifierStartAction(new DehumidifierModule(runner, haOptions),
|
|
runner, haOptions, TimeProvider.System);
|
|
var room = new IndoorEnvironmentReader(haOptions, TimeProvider.System);
|
|
var climate = new BedPresenceClimatePreview(gateway, room, haOptions);
|
|
var heat = new BedPresenceClimateHeatAction(new ClimateModule(runner, haOptions), room, runner, haOptions);
|
|
var fragrance = new BedPresenceFragranceReleaseAction(runner, haOptions, wakeDevice: false);
|
|
var wakeFragrance = new BedPresenceFragranceReleaseAction(runner, haOptions, wakeDevice: true);
|
|
var audio = new FakeAudioGateway();
|
|
var music = new TcmMusicTransitionService(audio, Options.Create(new TcmMusicOptions
|
|
{
|
|
Tracks = new() { ["balanced"] = [new() { Url = TrackUrl, DurationSeconds = 30 }] }
|
|
}), NullLogger<TcmMusicTransitionService>.Instance, new ManualTestTimeProvider());
|
|
var speaker = new BedPresenceSleepMusicAction(gateway, music, haOptions);
|
|
var executionLog = new RecordingExecutionLog();
|
|
using var automation = new BedPresenceAutomation(Options.Create(new BedPresenceAutomationOptions
|
|
{
|
|
Mode = BedAutomationMode.Managed, ActionKey = curtainAction.Key, TargetEntityId = TargetCurtain,
|
|
AdditionalRules = [new() { Mode = BedAutomationMode.Managed, ActionKey = preview.Key, TargetEntityId = TargetDehumidifier },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = dehumidifier.Key, TargetEntityId = TargetDehumidifier },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = climate.Key, TargetEntityId = TargetClimate },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = heat.Key, TargetEntityId = TargetClimate },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = fragrance.Key, TargetEntityId = FragranceButton },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = wakeFragrance.Key, TargetEntityId = WakeFragranceButton },
|
|
new() { Mode = BedAutomationMode.Managed, ActionKey = speaker.Key, TargetEntityId = Speaker }]
|
|
}), [curtainAction, preview, dehumidifier, climate, heat, fragrance, wakeFragrance, speaker], TimeProvider.System,
|
|
NullLogger<BedPresenceAutomation>.Instance, executionLog);
|
|
var observedPresence = new ObservedPresenceSink(automation);
|
|
using var service = new Ib43MqttService(options, new Ib43MqttTransportFactory(), store,
|
|
TimeProvider.System, NullLogger<Ib43MqttService>.Instance, sink, observedPresence);
|
|
using var publisher = factory.CreateMqttClient();
|
|
try
|
|
{
|
|
await automation.StartAsync(default);
|
|
await service.StartAsync(default);
|
|
await WaitFor(() => store.Snapshot()[0].Transport == "connected");
|
|
await publisher.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", port)
|
|
.WithClientId("ib43-local-simulator").Build());
|
|
foreach (var name in new[] { "startup.pdf-example.json", "realtime.pdf-example.json", "sleep-event.pdf-keys.json" })
|
|
{
|
|
var json = JsonNode.Parse(File.ReadAllText(Path.Combine(AppContext.BaseDirectory, "Ib43Examples", name)));
|
|
json["params"]["deviceName"] = binding.SerialNumber;
|
|
if (json["params"]["productKey"] != null) json["params"]["productKey"] = binding.ProductKey;
|
|
await publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic)
|
|
.WithPayload(json.ToJsonString()).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
|
|
}
|
|
await WaitFor(() => store.Snapshot()[0].Readings.TryGetValue("left.heartRate", out var value) && value.Value == 74);
|
|
Assert.Equal(80d, store.Snapshot()[0].Readings["right.heartRate"].Value);
|
|
Assert.Equal(60, store.Snapshot()[0].Fields["characteristic"].Value.GetProperty("reportIerVal").GetInt32());
|
|
Assert.All(store.Diagnostics(), entry => Assert.Empty(entry.Issues));
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 2);
|
|
Assert.Equal(74d, (await sink.Repository.Latest(1, LeftBinding))["heartRate"].Value);
|
|
Assert.Equal(80d, (await sink.Repository.Latest(2, "right-test"))["heartRate"].Value);
|
|
Assert.Empty(gateway.Calls);
|
|
|
|
await PublishPresence(publisher, binding, 0, 0);
|
|
await PublishPresence(publisher, binding, 1, 0);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 4);
|
|
Assert.False(automation.IsEnabled(LeftBinding, 1));
|
|
Assert.Empty(gateway.Calls);
|
|
var bed = new BedAutomationBinding(LeftBinding, 1, binding.ProductKey, binding.SerialNumber, BedSide.Left, true);
|
|
Assert.True(automation.TryEnable(bed));
|
|
|
|
// 左侧空床后上床才触发。连续重复在床不会重启，右侧的变化不触发左侧规则。
|
|
await PublishPresence(publisher, binding, 0, 0);
|
|
await PublishPresence(publisher, binding, 0, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 8 && gateway.Calls.Count == 1);
|
|
Assert.Equal("readback-confirmed", await executionLog.WaitForResult(curtainAction.Key));
|
|
var close = Assert.Single(gateway.Calls);
|
|
Assert.Equal(TargetCurtain, close.EntityId);
|
|
Assert.Equal("close_cover", close.Service);
|
|
Assert.Equal("closed", gateway.States[TargetCurtain].State);
|
|
Assert.Equal(1d, (await sink.Repository.Latest(1, LeftBinding))["presence"].Value);
|
|
automation.Disable(LeftBinding, 1);
|
|
|
|
Assert.True(automation.TryEnable(bed, preview.Key));
|
|
Assert.False(automation.IsEnabled(LeftBinding, 1));
|
|
await PublishPresence(publisher, binding, 0, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 11);
|
|
Assert.Equal("preview-would-start", await executionLog.WaitForResult(preview.Key));
|
|
Assert.Single(gateway.Calls);
|
|
Assert.Equal("off", gateway.States[TargetDehumidifier].State);
|
|
Assert.True(automation.Disable(LeftBinding, 1, preview.Key));
|
|
|
|
Assert.True(automation.TryEnable(bed, dehumidifier.Key));
|
|
await PublishPresence(publisher, binding, 0, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await PublishPresence(publisher, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 14);
|
|
Assert.Equal("readback-confirmed", await executionLog.WaitForResult(dehumidifier.Key));
|
|
Assert.Equal("on", gateway.States[TargetDehumidifier].State);
|
|
|
|
gateway.States[TargetCurtain] = openCurtain;
|
|
gateway.States[TargetDehumidifier] = stoppedDehumidifier;
|
|
Assert.True(automation.TryEnable(bed));
|
|
await PublishPresence(publisher, binding, 0, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Empty });
|
|
var oldConnection = observedPresence.LastLeft!.ConnectionId;
|
|
|
|
// 保持两条实物规则开启，断开前的空床基线不能与新连接的首条在床组成上床事件。
|
|
await broker.StopAsync();
|
|
await WaitFor(() => store.Snapshot()[0].Transport != "connected");
|
|
await broker.StartAsync();
|
|
await WaitFor(() => store.Snapshot()[0].Transport == "connected");
|
|
using var publisherAfterRestart = factory.CreateMqttClient();
|
|
await publisherAfterRestart.ConnectAsync(new MqttClientOptionsBuilder().WithTcpServer("127.0.0.1", port).Build());
|
|
Assert.True(automation.IsEnabled(LeftBinding, 1));
|
|
Assert.True(automation.IsEnabled(LeftBinding, 1, dehumidifier.Key));
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Occupied });
|
|
var firstOnNewConnection = observedPresence.LastLeft!;
|
|
Assert.NotEqual(oldConnection, firstOnNewConnection.ConnectionId);
|
|
Assert.Equal(3, executionLog.Started.Count);
|
|
Assert.Equal(2, gateway.Calls.Count);
|
|
|
|
await PublishPresence(publisherAfterRestart, binding, 0, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Empty });
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Occupied });
|
|
var freshTransition = observedPresence.LastLeft!;
|
|
Assert.Equal(firstOnNewConnection.ConnectionId, freshTransition.ConnectionId);
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 19 && executionLog.Completed.Count >= 5);
|
|
|
|
await publisherAfterRestart.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic)
|
|
.WithPayload("{\"method\":\"thing.property.post\",\"params\":{\"deviceName\":\"IB43SIM001\",\"HR\":[75,16,0,81,18,0]}}")
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
|
|
await WaitFor(() => store.Snapshot()[0].Readings["left.heartRate"].Value == 75);
|
|
Assert.Equal(1d, store.Snapshot()[0].Readings["left.presence"].Value);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 20);
|
|
Assert.Equal(75d, (await sink.Repository.Latest(1, LeftBinding))["heartRate"].Value);
|
|
var newExecutions = executionLog.Started.Skip(3).ToArray();
|
|
Assert.Equal(new[] { curtainAction.Key, dehumidifier.Key },
|
|
newExecutions.Select(entry => entry.Action).OrderBy(key => key, StringComparer.Ordinal));
|
|
Assert.All(newExecutions, entry => Assert.Equal(freshTransition.ReceivedAt, entry.TriggeredAt));
|
|
Assert.All(executionLog.Completed.Skip(3), result => Assert.Equal("readback-confirmed", result));
|
|
Assert.Equal(5, executionLog.Completed.Count);
|
|
Assert.Equal(new[] { "close_cover", "turn_on" },
|
|
gateway.Calls.Skip(2).Select(command => command.Service).OrderBy(name => name, StringComparer.Ordinal));
|
|
|
|
Assert.True(automation.Disable(LeftBinding, 1));
|
|
Assert.True(automation.Disable(LeftBinding, 1, dehumidifier.Key));
|
|
Assert.False(automation.IsEnabled(LeftBinding, 1, climate.Key));
|
|
Assert.True(automation.TryEnable(bed, climate.Key));
|
|
await PublishPresence(publisherAfterRestart, binding, 0, 1);
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 23);
|
|
Assert.Equal("preview-climate-cool", await executionLog.WaitForResult(climate.Key));
|
|
Assert.Equal(climate.Key, Assert.Single(executionLog.Started.Skip(5)).Action);
|
|
Assert.Equal(4, gateway.Calls.Count);
|
|
|
|
Assert.True(automation.Disable(LeftBinding, 1, climate.Key));
|
|
Assert.False(automation.IsEnabled(LeftBinding, 1, heat.Key));
|
|
gateway.States[TargetCurtain] = openCurtain;
|
|
gateway.States[TargetDehumidifier] = stoppedDehumidifier;
|
|
gateway.States[TemperatureSensor] = new() { EntityId = TemperatureSensor, State = "20", LastReported = DateTimeOffset.UtcNow,
|
|
Attributes = JsonSerializer.SerializeToElement(new { unit_of_measurement = "°C" }) };
|
|
gateway.States[TargetClimate] = new() { EntityId = TargetClimate, State = "off",
|
|
Attributes = JsonSerializer.SerializeToElement(new { hvac_modes = new[] { "off", "heat", "cool" }, supported_features = 1,
|
|
temperature_unit = "°C", min_temp = 16, max_temp = 30, target_temp_step = 1, temperature = 24 }) };
|
|
Assert.True(automation.TryEnable(bed));
|
|
Assert.True(automation.TryEnable(bed, dehumidifier.Key));
|
|
Assert.True(automation.TryEnable(bed, heat.Key));
|
|
foreach (var action in new[] { fragrance.Key, wakeFragrance.Key, speaker.Key })
|
|
{
|
|
Assert.False(automation.IsEnabled(LeftBinding, 1, action));
|
|
Assert.True(automation.TryEnable(bed, action));
|
|
}
|
|
await PublishPresence(publisherAfterRestart, binding, 0, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Empty });
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => observedPresence.LastLeft is { Presence: BedPresence.Occupied });
|
|
var combinedTransition = observedPresence.LastLeft!;
|
|
await PublishPresence(publisherAfterRestart, binding, 1, 1);
|
|
await WaitFor(() => Volatile.Read(ref sink.SavedReports) == 26 && executionLog.Completed.Count >= 12);
|
|
var combinedExecutions = executionLog.Started.Skip(6).ToArray();
|
|
Assert.Equal(new[] { heat.Key, curtainAction.Key, dehumidifier.Key, fragrance.Key, speaker.Key, wakeFragrance.Key },
|
|
combinedExecutions.Select(entry => entry.Action).OrderBy(key => key, StringComparer.Ordinal));
|
|
Assert.All(combinedExecutions, entry => Assert.Equal(combinedTransition.ReceivedAt, entry.TriggeredAt));
|
|
var combinedResults = executionLog.Completed.Skip(6).ToArray();
|
|
Assert.Equal(6, combinedResults.Length);
|
|
Assert.Equal(3, combinedResults.Count(result => result == "readback-confirmed"));
|
|
Assert.Equal("fragrance-release-sent", await executionLog.WaitForResult(fragrance.Key));
|
|
Assert.Equal("fragrance-release-sent", await executionLog.WaitForResult(wakeFragrance.Key));
|
|
Assert.Equal("speaker-playback-sent", await executionLog.WaitForResult(speaker.Key));
|
|
|
|
var combinedCommands = gateway.Calls.Skip(4).ToArray();
|
|
Assert.Equal(new[] { "close_cover", "press", "press", "set_hvac_mode", "set_temperature", "turn_on" },
|
|
combinedCommands.Select(command => command.Service).OrderBy(name => name, StringComparer.Ordinal));
|
|
Assert.Equal(new[] { FragranceButton, WakeFragranceButton }, combinedCommands.Where(command => command.Service == "press")
|
|
.Select(command => command.EntityId).OrderBy(id => id, StringComparer.Ordinal));
|
|
Assert.Equal(TrackUrl, Assert.Single(audio.Tracks));
|
|
Assert.Equal(50, Assert.Single(audio.Volumes));
|
|
Assert.Equal(new[] { "set_hvac_mode", "set_temperature" },
|
|
combinedCommands.Where(command => command.EntityId == TargetClimate).Select(command => command.Service));
|
|
Assert.Equal("closed", gateway.States[TargetCurtain].State);
|
|
Assert.Equal("on", gateway.States[TargetDehumidifier].State);
|
|
Assert.Equal("heat", gateway.States[TargetClimate].State);
|
|
Assert.Equal(22, gateway.States[TargetClimate].GetDoubleAttribute("temperature"));
|
|
|
|
Assert.Equal("speaker-not-idle", await speaker.ExecuteAsync(Speaker, default));
|
|
Assert.Single(audio.Tracks);
|
|
gateway.States[WakeFragranceButton] = new() { EntityId = WakeFragranceButton, State = "unavailable" };
|
|
Assert.Equal("fragrance-release-unavailable", await wakeFragrance.ExecuteAsync(WakeFragranceButton, default));
|
|
Assert.Equal("target-not-configured", await fragrance.ExecuteAsync(WakeFragranceButton, default));
|
|
Assert.Equal(10, gateway.Calls.Count);
|
|
}
|
|
finally
|
|
{
|
|
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
|
await service.StopAsync(timeout.Token);
|
|
await automation.StopAsync(timeout.Token);
|
|
await music.StopAsync(timeout.Token);
|
|
await broker.StopAsync();
|
|
}
|
|
}
|
|
|
|
private sealed class ObservedPresenceSink(IBedPresenceSink inner) : IBedPresenceSink
|
|
{
|
|
private BedPresenceSample? _lastLeft;
|
|
public BedPresenceSample? LastLeft => Volatile.Read(ref _lastLeft);
|
|
|
|
public void Observe(BedPresenceSample sample, CancellationToken connectionToken)
|
|
{
|
|
inner.Observe(sample, connectionToken);
|
|
if (sample.Side == BedSide.Left) Volatile.Write(ref _lastLeft, sample);
|
|
}
|
|
}
|
|
|
|
private sealed class RecordingExecutionLog : IBedPresenceExecutionLog
|
|
{
|
|
private readonly ConcurrentDictionary<string, TaskCompletionSource<string>> _results = new();
|
|
public ConcurrentQueue<(string Action, DateTimeOffset TriggeredAt)> Started { get; } = new();
|
|
public ConcurrentQueue<string> Completed { get; } = new();
|
|
public Task<string> WaitForResult(string action) => Result(action).Task.WaitAsync(TimeSpan.FromSeconds(3));
|
|
|
|
private TaskCompletionSource<string> Result(string action) =>
|
|
_results.GetOrAdd(action, _ => new(TaskCreationOptions.RunContinuationsAsynchronously));
|
|
public Task<string?> BeginAsync(string bindingId, long userId, string action, string target, DateTimeOffset triggeredAt, CancellationToken token)
|
|
{
|
|
Assert.Equal(LeftBinding, bindingId);
|
|
Assert.Equal(1L, userId);
|
|
Started.Enqueue((action, triggeredAt));
|
|
return Task.FromResult<string?>(action);
|
|
}
|
|
public Task CompleteAsync(string id, string result, DateTimeOffset completedAt)
|
|
{
|
|
Completed.Enqueue(result);
|
|
Result(id).TrySetResult(result);
|
|
return Task.CompletedTask;
|
|
}
|
|
}
|
|
|
|
private sealed class DatabaseSink : IIb43ReportSink, IDisposable
|
|
{
|
|
private readonly SqlSugarClient _db;
|
|
public readonly Ib43DataRepository Repository;
|
|
public int SavedReports;
|
|
public DatabaseSink(Ib43Binding binding)
|
|
{
|
|
_db = new(new ConnectionConfig { ConnectionString = "DataSource=:memory:", DbType = DbType.Sqlite,
|
|
IsAutoCloseConnection = false, InitKeyType = InitKeyType.Attribute,
|
|
ConfigureExternalServices = new ConfigureExternalServices { EntityService = (p, c) =>
|
|
{
|
|
if (p.DeclaringType == typeof(HealthIb43Observation) && p.Name == "Id") c.DataType = "INTEGER";
|
|
} } });
|
|
_db.CodeFirst.InitTables<HealthIb43Binding, HealthIb43Observation>();
|
|
Repository = new(_db);
|
|
var key = Ib43DataRepository.DeviceKey(binding.ProductKey, binding.SerialNumber);
|
|
foreach (var (side, user) in new[] { ("left", 1L), ("right", 2L) })
|
|
_db.Insertable(new HealthIb43Binding { Id = side == "left" ? LeftBinding : "right-test", DeviceKey = key,
|
|
ProductKey = binding.ProductKey, SerialNumber = binding.SerialNumber, Side = side,
|
|
ActiveSlot = key + "." + side, UserId = user, Simulation = true }).ExecuteCommand();
|
|
}
|
|
public async Task<Ib43SavedReport> SaveAsync(string reportId, Ib43Binding binding, Ib43Report report, CancellationToken token)
|
|
{
|
|
var count = await Repository.Save(reportId, binding, report);
|
|
Interlocked.Increment(ref SavedReports);
|
|
return count;
|
|
}
|
|
public void Dispose() => _db.Dispose();
|
|
}
|
|
|
|
private static Task<MqttClientPublishResult> PublishPresence(IMqttClient publisher, Ib43Binding binding, int left, int right)
|
|
{
|
|
var payload = JsonSerializer.Serialize(new
|
|
{
|
|
method = "thing.property.post",
|
|
@params = new { deviceName = binding.SerialNumber, SleepReportNew = new { ibNew = new[] { left, 0, 0, 16, 74, 0, right, 0, 0, 17, 80, 0 } } }
|
|
});
|
|
return publisher.PublishAsync(new MqttApplicationMessageBuilder().WithTopic(binding.PostTopic).WithPayload(payload)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce).Build());
|
|
}
|
|
|
|
private static async Task WaitFor(Func<bool> predicate)
|
|
{
|
|
using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(12));
|
|
while (!predicate()) await Task.Delay(20, timeout.Token);
|
|
}
|
|
}
|