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

Allow to disable EventPriority

Browse source
This commit is contained in:
Haiping Chen 2024-11-23 14:48:22 +00:00
parent 11ceb8e013
commit 597e26a1e0
4 changed files with 32 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs
View file

@ -10,7 +10,7 @@ public interface IEventPublisher
/// <returns></returns>
Task BroadcastAsync(string channel, string message);
Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium);
Task PublishAsync(string channel, string message, EventPriority? priority = null);
Task ReDispatchAsync(string channel, int count = 10, string order = "asc");

2
src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs
View file

@ -4,5 +4,5 @@ public interface IEventSubscriber
{
Task SubscribeAsync(string channel, Func<string, string, Task> received);
Task SubscribeAsync(string channel, string group, Func<string, string, Task> received);
Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func<string, string, Task> received);
}

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

@ -21,12 +21,15 @@ public class RedisPublisher : IEventPublisher
await _subscriber.PublishAsync(channel, message);
}
public async Task PublishAsync(string channel, string message, EventPriority priority = EventPriority.Medium)
public async Task PublishAsync(string channel, string message, EventPriority? priority = null)
{
var db = _redis.GetDatabase();
// convert to apporiate channel by priority
channel = $"{channel}-{priority}";
if (priority != null)
{
channel = $"{channel}-{priority}";
}
if (CheckMessageExists(db, channel, "message", message))
{

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

@ -25,27 +25,41 @@ public class RedisSubscriber : IEventSubscriber
});
}
public async Task SubscribeAsync(string channel, string group, Func<string, string, Task> received)
public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func<string, string, Task> received)
{
var db = _redis.GetDatabase();
await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group);
await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group);
await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group);
if (priorityEnabled)
{
await CreateConsumerGroup(db, $"{channel}-{EventPriority.Low}", group);
await CreateConsumerGroup(db, $"{channel}-{EventPriority.Medium}", group);
await CreateConsumerGroup(db, $"{channel}-{EventPriority.High}", group);
}
else
{
await CreateConsumerGroup(db, channel, group);
}
while (true)
{
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0)
if (priorityEnabled)
{
continue;
}
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0)
{
continue;
}
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0)
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0)
{
continue;
}
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received);
}
else
{
continue;
await HandleGroupMessage(db, channel, group, received);
}
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received);
}
}