Allow Event to be removed.
This commit is contained in:
parent
4813c5b0f4
commit
5f31ffe658
|
|
@ -15,4 +15,6 @@ public interface IEventPublisher
|
|||
Task ReDispatchAsync(string channel, int count = 10, string order = "asc");
|
||||
|
||||
Task ReDispatchPendingAsync(string channel, string group, int count = 10);
|
||||
|
||||
Task RemoveAsync(string channel, int count = 10);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -158,4 +158,26 @@ public class RedisPublisher : IEventPublisher
|
|||
Console.WriteLine($"Redis error: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
public async Task RemoveAsync(string channel, int count = 10)
|
||||
{
|
||||
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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue