79 lines
2.9 KiB
C#
79 lines
2.9 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Threading.Tasks;
|
|
using EasyNetQ;
|
|
using EasyNetQ.Topology;
|
|
using RabbitMQHelper.Models;
|
|
using Microsoft.AspNetCore.Http;
|
|
using Microsoft.Extensions.Logging;
|
|
using MongoDB.Driver;
|
|
|
|
namespace RabbitMQHelper
|
|
{
|
|
public class SendMQMessage:ISendMQMessage
|
|
{
|
|
private readonly IBus _bus;
|
|
private readonly ILogger<SendMQMessage> _logger;
|
|
private readonly IRabbitConfig _mongoDBConfig;
|
|
private readonly IMongoCollection<PublishRecord> _collection;
|
|
private readonly IHttpContextAccessor _httpContextAccessor;
|
|
|
|
public SendMQMessage(IBus bus, ILogger<SendMQMessage> logger, IRabbitConfig mongoDBConfig, IHttpContextAccessor httpContextAccessor)
|
|
{
|
|
_bus = bus;
|
|
_logger = logger;
|
|
_mongoDBConfig = mongoDBConfig;
|
|
_httpContextAccessor = httpContextAccessor;
|
|
var database = new MongoClient(_mongoDBConfig.MongoConnectionString).GetDatabase(_mongoDBConfig.MongoDatabaseName);
|
|
_collection = database.GetCollection<PublishRecord>("mq_publish_record");
|
|
}
|
|
public async Task<bool> Send(string topic, object value)
|
|
{
|
|
if (!_bus.Advanced.IsConnected)
|
|
{
|
|
|
|
return false;
|
|
}
|
|
Dictionary<string, string> headers = new Dictionary<string, string>();
|
|
|
|
if (_httpContextAccessor.HttpContext != null && _httpContextAccessor.HttpContext.Request != null && _httpContextAccessor.HttpContext.Request.Headers != null)
|
|
{
|
|
if (_httpContextAccessor.HttpContext.Request.Headers.ContainsKey("token"))
|
|
{
|
|
headers.Add("token", _httpContextAccessor.HttpContext.Request.Headers["token"].ToString() ?? "");
|
|
}
|
|
|
|
;
|
|
}
|
|
bool success = false;
|
|
|
|
var model = new RabbitMessage(value, headers);
|
|
|
|
var exchange = await _bus.Advanced.ExchangeDeclareAsync(_mongoDBConfig.DefaultExchange, ExchangeType.Topic);
|
|
|
|
var message = new Message<RabbitMessage>(model);
|
|
await _bus.Advanced.PublishAsync<RabbitMessage>(exchange, topic, false, message).ContinueWith(m =>
|
|
{
|
|
if (m.IsCompleted && m.IsCompletedSuccessfully)
|
|
{
|
|
success = true;
|
|
}
|
|
else
|
|
{
|
|
_logger.LogError(m.Exception, "RabbitMQ消息推送失败 topic={Topic}", topic);
|
|
}
|
|
});
|
|
try
|
|
{
|
|
await _collection.InsertOneAsync(new PublishRecord(model.RecordId, topic, value, success));
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "RabbitMQ推送记录写入异常 topic={Topic}", topic);
|
|
}
|
|
|
|
return success;
|
|
}
|
|
}
|
|
}
|