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

Support port for event

Browse source
This commit is contained in:
Haiping Chen 2024-12-01 23:06:21 +00:00
parent 4cf3531c32
commit 3dc76fc7f5
2 changed files with 14 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -6,7 +6,7 @@ public interface IEventSubscriber
{
Task SubscribeAsync(string channel, Func<string, string, Task> received);
Task SubscribeAsync(string channel, string group, bool priorityEnabled,
Task SubscribeAsync(string channel, string group, int? port, bool priorityEnabled,
Func<string, string, Task> received,
CancellationToken? stoppingToken = null);
}

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

@ -25,7 +25,7 @@ public class RedisSubscriber : IEventSubscriber
});
}
public async Task SubscribeAsync(string channel, string group, bool priorityEnabled,
public async Task SubscribeAsync(string channel, string group, int? port, bool priorityEnabled,
Func<string, string, Task> received,
CancellationToken? stoppingToken = null)
{
@ -42,6 +42,12 @@ public class RedisSubscriber : IEventSubscriber
await CreateConsumerGroup(db, channel, group);
}
var consumer = Environment.MachineName;
if (port.HasValue)
{
consumer += $"-{port}";
}
while (true)
{
await Task.Delay(100);
@ -54,28 +60,28 @@ public class RedisSubscriber : IEventSubscriber
if (priorityEnabled)
{
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0)
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, consumer, received) > 0)
{
continue;
}
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, received) > 0)
if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received) > 0)
{
continue;
}
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, received);
await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received);
}
else
{
await HandleGroupMessage(db, channel, group, received);
await HandleGroupMessage(db, channel, group, consumer, received);
}
}
}
private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, Func<string, string, Task> received)
private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, string consumer, Func<string, string, Task> received)
{
var entries = await db.StreamReadGroupAsync(channel, group, Environment.MachineName, count: 1);
var entries = await db.StreamReadGroupAsync(channel, group, consumer, count: 1);
foreach (var entry in entries)
{
_logger.LogInformation($"Consumer {Environment.MachineName} received: {channel} {entry.Values[0].Value}");