Skip to content

Commit 449fc11

Browse files
hhvrchhvrc
committed
Finalize implementation
1 parent 8ab7104 commit 449fc11

3 files changed

Lines changed: 75 additions & 25 deletions

File tree

‎API/Realtime/RedisSubscriberService.cs‎

Lines changed: 58 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,18 @@
1-
using System.Text.Json;
1+
using Google.Protobuf.WellKnownTypes;
2+
using MessagePack;
23
using Microsoft.AspNetCore.SignalR;
34
using Microsoft.EntityFrameworkCore;
45
using OpenShock.Common.Hubs;
56
using OpenShock.Common.Models.WebSocket;
67
using OpenShock.Common.OpenShockDb;
78
using OpenShock.Common.Redis;
9+
using OpenShock.Common.Redis.PubSub;
810
using OpenShock.Common.Services.RedisPubSub;
911
using OpenShock.Common.Utils;
1012
using Redis.OM.Contracts;
1113
using StackExchange.Redis;
14+
using System.Text.Json;
15+
using static System.Runtime.InteropServices.JavaScript.JSType;
1216

1317
namespace OpenShock.API.Realtime;
1418

@@ -21,6 +25,7 @@ public sealed class RedisSubscriberService : IHostedService, IAsyncDisposable
2125
private readonly IDbContextFactory<OpenShockContext> _dbContextFactory;
2226
private readonly IRedisConnectionProvider _redisConnectionProvider;
2327
private readonly ISubscriber _subscriber;
28+
private readonly ILogger<RedisSubscriberService> _logger;
2429

2530
/// <summary>
2631
/// DI Constructor
@@ -29,34 +34,78 @@ public sealed class RedisSubscriberService : IHostedService, IAsyncDisposable
2934
/// <param name="hubContext"></param>
3035
/// <param name="dbContextFactory"></param>
3136
/// <param name="redisConnectionProvider"></param>
37+
/// <param name="logger"></param>
3238
public RedisSubscriberService(
3339
IConnectionMultiplexer connectionMultiplexer,
3440
IHubContext<UserHub, IUserHub> hubContext,
3541
IDbContextFactory<OpenShockContext> dbContextFactory,
36-
IRedisConnectionProvider redisConnectionProvider)
42+
IRedisConnectionProvider redisConnectionProvider,
43+
ILogger<RedisSubscriberService> logger
44+
)
3745
{
3846
_hubContext = hubContext;
3947
_dbContextFactory = dbContextFactory;
4048
_redisConnectionProvider = redisConnectionProvider;
4149
_subscriber = connectionMultiplexer.GetSubscriber();
50+
_logger = logger;
4251
}
4352

4453
/// <inheritdoc />
4554
public async Task StartAsync(CancellationToken cancellationToken)
4655
{
4756
await _subscriber.SubscribeAsync(RedisChannels.KeyEventExpired, (_, message) => OsTask.Run(() => HandleKeyExpired(message)));
48-
await _subscriber.SubscribeAsync(RedisChannels.DeviceStatus, (_, message) => OsTask.Run(() => HandleDeviceStatus(message)));
57+
await _subscriber.SubscribeAsync(RedisChannels.DeviceStatus, ProcessDeviceStatusEvent);
4958
}
5059

51-
private async Task HandleDeviceStatus(RedisValue message)
60+
private void ProcessDeviceStatusEvent(RedisChannel _, RedisValue value)
5261
{
53-
if (!message.HasValue) return;
54-
var data = JsonSerializer.Deserialize<DeviceUpdatedMessage>(message.ToString());
55-
if (data is null) return;
62+
if (!value.HasValue) return;
63+
64+
DeviceStatus message;
65+
try
66+
{
67+
message = MessagePackSerializer.Deserialize<DeviceStatus>((ReadOnlyMemory<byte>)value);
68+
if (message is null) return;
69+
}
70+
catch (Exception e)
71+
{
72+
_logger.LogError(e, "Failed to deserialize redis message");
73+
return;
74+
}
5675

57-
await LogicDeviceOnlineStatus(data.Id);
76+
OsTask.Run(() => HandleDeviceStatusMessage(message));
5877
}
59-
78+
79+
private async Task HandleDeviceStatusMessage(DeviceStatus message)
80+
{
81+
switch (message.Payload)
82+
{
83+
case DeviceBoolStatePayload boolState:
84+
await HandleDeviceBoolState(message.DeviceId, boolState);
85+
break;
86+
default:
87+
_logger.LogError("Got DeviceStatus with unknown payload type: {PayloadType}", message.Payload?.GetType().Name);
88+
break;
89+
}
90+
91+
}
92+
93+
private async Task HandleDeviceBoolState(Guid deviceId, DeviceBoolStatePayload state)
94+
{
95+
switch (state.Type)
96+
{
97+
case DeviceBoolStateType.Online:
98+
await LogicDeviceOnlineStatus(deviceId); // TODO: Handle device offline messages too
99+
break;
100+
case DeviceBoolStateType.EStopped:
101+
_logger.LogWarning("This is not yet implemented");
102+
break;
103+
default:
104+
_logger.LogError("Unknown DeviceBoolStateType: {StateType}", state.Type);
105+
break;
106+
}
107+
}
108+
60109
private async Task HandleKeyExpired(RedisValue message)
61110
{
62111
if (!message.HasValue) return;

‎Common/Redis/PubSub/DeviceStatus.cs‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,12 @@ namespace OpenShock.Common.Redis.PubSub;
55
[MessagePackObject]
66
public sealed class DeviceStatus
77
{
8-
[Key(0)] public DeviceStatusType Type { get; init; }
8+
[Key(0)] public required Guid DeviceId { get; init; }
99
[Key(1)] public required IDeviceStatusPayload Payload { get; init; }
1010

11-
public static DeviceStatus Create(DeviceBoolStateType stateType, bool state) => new()
11+
public static DeviceStatus Create(Guid deviceId, DeviceBoolStateType stateType, bool state) => new()
1212
{
13-
Type = DeviceStatusType.BoolStateChanged,
13+
DeviceId = deviceId,
1414
Payload = new DeviceBoolStatePayload
1515
{
1616
Type = stateType,
@@ -19,11 +19,6 @@ public sealed class DeviceStatus
1919
};
2020
}
2121

22-
public enum DeviceStatusType : byte
23-
{
24-
BoolStateChanged = 0,
25-
}
26-
2722
[Union(0, typeof(DeviceBoolStatePayload))]
2823
public interface IDeviceStatusPayload;
2924

‎Common/Services/RedisPubSub/RedisPubService.cs‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -14,40 +14,46 @@ public RedisPubService(IConnectionMultiplexer connectionMultiplexer)
1414
_subscriber = connectionMultiplexer.GetSubscriber();
1515
}
1616

17-
private Task<long> Publish<T>(Guid deviceId, T msg) => _subscriber.PublishAsync(RedisChannels.DeviceMessage(deviceId), (RedisValue)new ReadOnlyMemory<byte>(MessagePackSerializer.Serialize(msg)));
17+
private Task<long> Publish<T>(RedisChannel channel, T msg) => _subscriber.PublishAsync(channel, (RedisValue)new ReadOnlyMemory<byte>(MessagePackSerializer.Serialize(msg)));
18+
private Task<long> PublishMessage(Guid deviceId, DeviceMessage msg) => Publish(RedisChannels.DeviceMessage(deviceId), msg);
1819

1920
public Task SendDeviceUpdate(Guid deviceId)
2021
{
21-
return Publish(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceInfoUpdated));
22+
return PublishMessage(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceInfoUpdated));
2223
}
2324

2425
public Task SendDeviceOnlineStatus(Guid deviceId, bool isOnline)
2526
{
26-
return Publish(deviceId, DeviceStatus.Create(DeviceBoolStateType.Online, isOnline));
27+
return Publish(RedisChannels.DeviceStatus, DeviceStatus.Create(deviceId, DeviceBoolStateType.Online, isOnline));
28+
}
29+
30+
public Task SendDeviceEstoppedStatus(Guid deviceId, bool isEstopped)
31+
{
32+
return Publish(RedisChannels.DeviceStatus, DeviceStatus.Create(deviceId, DeviceBoolStateType.EStopped, isEstopped));
2733
}
2834

2935
public Task SendDeviceControl(Guid deviceId, List<DeviceControlPayload.ShockerControlInfo> controls)
3036
{
31-
return Publish(deviceId, DeviceMessage.Create(new DeviceControlPayload { Controls = controls }));
37+
return PublishMessage(deviceId, DeviceMessage.Create(new DeviceControlPayload { Controls = controls }));
3238
}
3339

3440
public Task SendDeviceCaptivePortal(Guid deviceId, bool enabled)
3541
{
36-
return Publish(deviceId, DeviceMessage.Create(DeviceToggleTarget.CaptivePortal, enabled));
42+
return PublishMessage(deviceId, DeviceMessage.Create(DeviceToggleTarget.CaptivePortal, enabled));
3743
}
3844

3945
public Task SendDeviceEmergencyStop(Guid deviceId)
4046
{
41-
return Publish(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceEmergencyStop));
47+
return PublishMessage(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceEmergencyStop));
4248
}
4349

4450
public Task SendDeviceOtaInstall(Guid deviceId, SemVersion version)
4551
{
46-
return Publish(deviceId, DeviceMessage.Create(new DeviceOtaInstallPayload { Version = version }));
52+
return PublishMessage(deviceId, DeviceMessage.Create(new DeviceOtaInstallPayload { Version = version }));
4753
}
4854

4955
public Task SendDeviceReboot(Guid deviceId)
5056
{
51-
return Publish(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceReboot));
57+
return PublishMessage(deviceId, DeviceMessage.Create(DeviceTriggerType.DeviceReboot));
5258
}
5359
}

0 commit comments

Comments
 (0)