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

Batch remove message from Redis stream

Browse source
This commit is contained in:
Haiping Chen 2024-12-31 17:07:38 +00:00
parent 6b6467c1d4
commit 1a8fa0b213
6 changed files with 5 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/Infrastructure/BotSharp.Core/BotSharp.Core.csproj
View file

@ -190,7 +190,7 @@
<ItemGroup>
<PackageReference Include="Aspects.Cache" Version="2.0.4" />
<PackageReference Include="DistributedLock.Redis" Version="1.0.3" />
<PackageReference Include="EntityFrameworkCore.BootKit" Version="8.7.1" />
<PackageReference Include="EntityFrameworkCore.BootKit" Version="8.8.0" />
<PackageReference Include="Fluid.Core" Version="2.11.1" />
<PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.1" />
<PackageReference Include="Microsoft.Extensions.Http" Version="8.0.1" />

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

@ -180,20 +180,7 @@ public class RedisPublisher : IEventPublisher
var db = _redis.GetDatabase();
var entries = await db.StreamRangeAsync(channel, "-", "+", count: count, messageOrder: Order.Ascending);
foreach (var entry in entries)
{
_logger.LogInformation($"Fetched message: {channel} {entry.Values[0].Value} ({entry.Id})");
try
{
await db.StreamDeleteAsync(channel, [entry.Id]);
_logger.LogWarning($"Deleted message: {channel} {entry.Values[0].Value} ({entry.Id})");
}
catch (Exception ex)
{
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
}
}
var deletedCount = await db.StreamDeleteAsync(channel, entries.Select(x => x.Id).ToArray());
_logger.LogWarning($"Deleted {deletedCount} messages from Redis stream {channel}");
}
}

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

@ -116,7 +116,8 @@ public class RedisSubscriber : IEventSubscriber
}
finally
{
await db.StreamDeleteAsync(channel, [entry.Id]);
var deletedCount = await db.StreamDeleteAsync(channel, [entry.Id]);
_logger.LogInformation($"Handled message {entry.Id}: {deletedCount == 1}");
}
}

4
src/Plugins/BotSharp.Plugin.HttpHandler/BotSharp.Plugin.HttpHandler.csproj
View file

@ -33,10 +33,6 @@
</Content>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Http" Version="8.0.0" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\Infrastructure\BotSharp.Core\BotSharp.Core.csproj" />
</ItemGroup>

4
src/Plugins/BotSharp.Plugin.SqlDriver/Hooks/SqlDriverPlanningHook.cs
View file

@ -1,13 +1,9 @@
using BotSharp.Abstraction.Agents.Enums;
using BotSharp.Abstraction.Messaging.Enums;
using BotSharp.Abstraction.Messaging.Models.RichContent.Template;
using BotSharp.Abstraction.Messaging.Models.RichContent;
using BotSharp.Abstraction.Messaging;
using BotSharp.Abstraction.Planning;
using BotSharp.Abstraction.Routing;
using BotSharp.Core.Infrastructures;
using System.Text.RegularExpressions;
using BotSharp.Plugin.SqlDriver.Interfaces;
namespace BotSharp.Plugin.SqlDriver.Hooks;

2
src/Plugins/BotSharp.Plugin.WebDriver/Functions/OpenBrowserFn.cs
View file

@ -1,5 +1,3 @@
using SQLitePCL;
namespace BotSharp.Plugin.WebDriver.Functions;
public class OpenBrowserFn : IFunctionCallback