70 lines
2.8 KiB
C#
70 lines
2.8 KiB
C#
using Fake.SyncEx;
|
||
using InterfaceForward.Application.Helpers;
|
||
using InterfaceForward.Application.Rabbit;
|
||
using InterfaceForward.Domain.Shared;
|
||
using InterfaceForward.Repositories;
|
||
using InterfaceForward.Repositories.App.Entitys;
|
||
using Microsoft.Extensions.Hosting;
|
||
using Newtonsoft.Json;
|
||
using RabbitMQ.Client;
|
||
|
||
namespace InterfaceForward.Application.HostServices;
|
||
|
||
public class AppSubscribeHostService : IHostedService
|
||
{
|
||
private readonly AppSubscribeEventHandler _appSubscribeEventHandler;
|
||
private readonly IBasicRepository<AppSubscribeConfigEntity> _appSubscribeConfigRepository;
|
||
private readonly RabbitClient _rabbitClient;
|
||
|
||
public AppSubscribeHostService(IBasicRepository<AppSubscribeConfigEntity> appSubscribeConfigRepository,
|
||
RabbitClient rabbitClient, AppSubscribeEventHandler appSubscribeEventHandler)
|
||
{
|
||
_appSubscribeConfigRepository = appSubscribeConfigRepository;
|
||
_rabbitClient = rabbitClient;
|
||
_appSubscribeEventHandler = appSubscribeEventHandler;
|
||
}
|
||
|
||
public async Task StartAsync(CancellationToken cancellationToken)
|
||
{
|
||
// todo：redis注册连接恢复订阅
|
||
RedisHelper.Client.Subscribe(AppSubscribeEventHandler.AppSubscribeEventSubscribe, AppSubScribe);
|
||
|
||
// 根据应用订阅表，绑定消费者
|
||
var subscribes = await _appSubscribeConfigRepository.GetListAsync(x => true);
|
||
foreach (var subscribe in subscribes)
|
||
{
|
||
var queueName = $"{BatchForwardEventHandler.BatchForwardEventSubscribe}.{subscribe.AppKey}.{subscribe
|
||
.SystemInterfaceCode}.{subscribe.ServiceProviderCode}";
|
||
_rabbitClient.Subscribe(new ConsumeOptions
|
||
{
|
||
Queue = queueName,
|
||
FetchCount = subscribe.FetchCount, // 并发控制
|
||
FailedRequeue = subscribe.FailedRequeue,
|
||
Declaration = o =>
|
||
{
|
||
o.ExchangeDeclare(GlobalConst.InterfaceRelayExchange, ExchangeType.Direct, true);
|
||
o.QueueDeclare(queueName, true);
|
||
o.QueueBind(queueName, GlobalConst.InterfaceRelayExchange, queueName);
|
||
}
|
||
});
|
||
}
|
||
}
|
||
|
||
public Task StopAsync(CancellationToken cancellationToken)
|
||
{
|
||
RedisHelper.Client.UnSubscribe(AppSubscribeEventHandler.AppSubscribeEventSubscribe);
|
||
return Task.CompletedTask;
|
||
}
|
||
|
||
public void AppSubScribe(string chan, object msg)
|
||
{
|
||
#if DEBUG
|
||
LogHelper.Info($"接受到来自redis的订阅：{chan}\n{msg.ToJson()}", true);
|
||
#endif
|
||
if (chan == AppSubscribeEventHandler.AppSubscribeEventSubscribe && msg is string str)
|
||
{
|
||
var @event = JsonConvert.DeserializeObject<AppSubscribeEvent>(str);
|
||
SyncContext.Run(() => _appSubscribeEventHandler.Handle(@event!));
|
||
}
|
||
}
|
||
} |