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

Add Redis timestamp

Browse source
This commit is contained in:
Haiping Chen 2024-11-18 22:31:18 +00:00
parent b01acea3a5
commit a14981e032
2 changed files with 8 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -24,7 +24,11 @@ public class RedisPublisher : IEventPublisher
{
var db = _redis.GetDatabase();
// Add a message to the stream, keeping only the latest 1 million messages
await db.StreamAddAsync(channel, "message", message,
await db.StreamAddAsync(channel,
[
new NameValueEntry("message", message),
new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o"))
],
maxLength: 1000 * 10000);
_logger.LogInformation($"Published message {channel} {message}");
@ -50,7 +54,7 @@ public class RedisPublisher : IEventPublisher
}
catch (Exception ex)
{
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}");
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
}
}
}

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

@ -50,6 +50,7 @@ public class RedisSubscriber : IEventSubscriber
foreach (var entry in entries)
{
_logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}");
await db.StreamAcknowledgeAsync(channel, group, entry.Id);
try
{
@ -60,11 +61,7 @@ public class RedisSubscriber : IEventSubscriber
}
catch (Exception ex)
{
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}");
}
finally
{
await db.StreamAcknowledgeAsync(channel, group, entry.Id);
_logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
}
}