SemianiakaVY/BotSharp
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Add Error channel

Browse source
This commit is contained in:
Haiping Chen 2024-12-27 18:16:23 +00:00
parent d041d42350
commit 7b14c8375d
2 changed files with 25 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs
View file

@ -79,7 +79,7 @@ public class RedisPublisher : IEventPublisher
return exists;
}
private NameValueEntry[] AssembleMessage(RedisValue message)
public static NameValueEntry[] AssembleMessage(RedisValue message)
{
return
[
@ -88,6 +88,16 @@ public class RedisPublisher : IEventPublisher
];
}
public static NameValueEntry[] AssembleErrorMessage(RedisValue message, string error)
{
return
[
new NameValueEntry("message", message),
new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")),
new NameValueEntry("error", error)
];
}
public async Task ReDispatchAsync(string channel, int count = 10, string order = "asc")
{
var db = _redis.GetDatabase();

18
src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs
View file

@ -42,6 +42,8 @@ public class RedisSubscriber : IEventSubscriber
await CreateConsumerGroup(db, channel, group);
}
await CreateConsumerGroup(db, $"{channel}-Error", group);
var consumer = Environment.MachineName;
if (port.HasValue)
{
@ -98,16 +100,24 @@ public class RedisSubscriber : IEventSubscriber
try
{
await received(channel, entry.Values[0].Value);
// Optionally delete the message to save space
await db.StreamDeleteAsync(channel, [entry.Id]);
}
catch (Exception ex)
{
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id} {entry.Values[0].Value}");
// Add a message to the Error stream, keeping only the latest 1 million messages
await db.StreamAddAsync($"{channel}-Error",
RedisPublisher.AssembleErrorMessage(entry.Values[0].Value, ex.Message),
messageId: entry.Id,
maxLength: 1000 * 10000);
// Slow down the consumer if there are errors
await Task.Delay(1000 * 10);
}
finally
{
await db.StreamDeleteAsync(channel, [entry.Id]);
}
}
return entries.Length;