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

Fix ErrorChannel

Browse source
This commit is contained in:
Haiping Chen 2024-12-29 16:23:31 +00:00
parent f2456caf35
commit 68df1f14c4
1 changed files with 6 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -64,21 +64,21 @@ public class RedisSubscriber : IEventSubscriber
{
if (priorityEnabled)
{
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, consumer, received) > 0)
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, consumer, received, $"{channel}-Error") > 0)
{
continue;
}
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received) > 0)
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received, $"{channel}-Error") > 0)
{
continue;
}
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received);
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received, $"{channel}-Error");
}
else
{
await HandleGroupMessage(db, channel, group, consumer, received);
await HandleGroupMessage(db, channel, group, consumer, received, $"{channel}-Error");
}
}
catch (Exception ex)
@ -89,7 +89,7 @@ public class RedisSubscriber : IEventSubscriber
}
}
private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, string consumer, Func<string, string, Task> received)
private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, string consumer, Func<string, string, Task> received, string errorChannel)
{
var entries = await db.StreamReadGroupAsync(channel, group, consumer, count: 1);
foreach (var entry in entries)
@ -106,7 +106,7 @@ public class RedisSubscriber : IEventSubscriber
_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",
await db.StreamAddAsync(errorChannel,
RedisPublisher.AssembleErrorMessage(entry.Values[0].Value, ex.Message),
messageId: entry.Id,
maxLength: 1000 * 10000);