birbnetes/birbmap
4
0
Fork 0
Code Pull Requests Releases Activity

Added RabbitMq support

Browse Source
This commit is contained in:
kunkliricsi
2021-01-16 15:23:10 +01:00
parent 265a59d4c3
commit 0df5b350d9
18 changed files with 274 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
Birdmap.BLL/Services/CommunationServices/CommunicationServiceBase.cs Normal file
View File

@ -0,0 +1,87 @@
using Birdmap.BLL.Interfaces;
using Birdmap.BLL.Services.CommunicationServices.Hubs;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Timer = System.Timers.Timer;
namespace Birdmap.BLL.Services.CommunationServices
{
internal class Payload
{
[JsonProperty("tag")]
public Guid TagID { get; set; }
[JsonProperty("probability")]
public double Probability { get; set; }
}
internal abstract class CommunicationServiceBase : ICommunicationService
{
protected readonly ILogger _logger;
protected readonly IInputService _inputService;
protected readonly IHubContext<DevicesHub, IDevicesHubClient> _hubContext;
private readonly Timer _hubTimer;
private readonly List<Message> _messages = new();
private readonly object _messageLock = new();
public abstract bool IsConnected { get; }
public CommunicationServiceBase(ILogger logger, IInputService inputService, IHubContext<DevicesHub, IDevicesHubClient> hubContext)
{
_logger = logger;
_inputService = inputService;
_hubContext = hubContext;
_hubTimer = new Timer()
{
AutoReset = true,
Interval = 1000,
};
_hubTimer.Elapsed += SendMqttMessagesWithSignalR;
}
protected async Task ProcessJsonMessageAsync(string json)
{
try
{
var payload = JsonConvert.DeserializeObject<Payload>(json);
var inputResponse = await _inputService.GetInputAsync(payload.TagID);
lock (_messageLock)
{
_messages.Add(new Message(inputResponse.Message.Device_id, inputResponse.Message.Date.UtcDateTime, payload.Probability));
}
}
catch (Exception ex)
{
_logger.LogError(ex, $"Could not handle application message.");
}
}
protected void StartMessageTimer()
{
_hubTimer.Start();
}
protected void StopMessageTimer()
{
_hubTimer.Stop();
}
private void SendMqttMessagesWithSignalR(object sender, System.Timers.ElapsedEventArgs e)
{
lock (_messageLock)
{
if (_messages.Any())
{
_logger.LogInformation($"Sending ({_messages.Count}) messages: {string.Join(" | ", _messages)}");
_hubContext.Clients.All.NotifyMessagesAsync(_messages);
_messages.Clear();
}
}
}
}
}

14
Birdmap.BLL/Services/CommunationServices/CommunicationServiceProvider.cs Normal file
View File

@ -0,0 +1,14 @@
using Birdmap.BLL.Interfaces;
namespace Birdmap.BLL.Services.CommunicationServices
{
internal class CommunicationServiceProvider : ICommunicationServiceProvider
{
public ICommunicationService Service { get; }
public CommunicationServiceProvider(ICommunicationService service)
{
Service = service;
}
}
}

67
Birdmap.BLL/Services/CommunationServices/Mqtt/MqttClientService.cs
View File

@ -1,4 +1,5 @@
using Birdmap.BLL.Interfaces;
using Birdmap.BLL.Services.CommunationServices;
using Birdmap.BLL.Services.CommunicationServices.Hubs;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
@ -7,59 +8,29 @@ using MQTTnet.Client;
using MQTTnet.Client.Connecting;
using MQTTnet.Client.Disconnecting;
using MQTTnet.Client.Options;
using Newtonsoft.Json;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Timer = System.Timers.Timer;
namespace Birdmap.BLL.Services.CommunicationServices.Mqtt
{
public class MqttClientService : IMqttClientService
internal class MqttClientService : CommunicationServiceBase, IMqttClientService
{
private readonly IMqttClient _mqttClient;
private readonly IMqttClientOptions _options;
private readonly ILogger<MqttClientService> _logger;
private readonly IInputService _inputService;
private readonly IHubContext<DevicesHub, IDevicesHubClient> _hubContext;
private readonly Timer _hubTimer;
private readonly List<Message> _messages = new();
private readonly object _messageLock = new();
public bool IsConnected => _mqttClient.IsConnected;
public override bool IsConnected => _mqttClient.IsConnected;
public MqttClientService(IMqttClientOptions options, ILogger<MqttClientService> logger, IInputService inputService, IHubContext<DevicesHub, IDevicesHubClient> hubContext)
: base(logger, inputService, hubContext)
{
_options = options;
_logger = logger;
_inputService = inputService;
_hubContext = hubContext;
_hubTimer = new Timer()
{
AutoReset = true,
Interval = 1000,
};
_hubTimer.Elapsed += SendMqttMessagesWithSignalR;
_mqttClient = new MqttFactory().CreateMqttClient();
ConfigureMqttClient();
}
private void SendMqttMessagesWithSignalR(object sender, System.Timers.ElapsedEventArgs e)
{
lock (_messageLock)
{
if (_messages.Any())
{
_logger.LogInformation($"Sending ({_messages.Count}) messages: {string.Join(" | ", _messages)}");
_hubContext.Clients.All.NotifyMessagesAsync(_messages);
_messages.Clear();
}
}
}
private void ConfigureMqttClient()
{
_mqttClient.ConnectedHandler = this;
@ -67,36 +38,14 @@ namespace Birdmap.BLL.Services.CommunicationServices.Mqtt
_mqttClient.ApplicationMessageReceivedHandler = this;
}
private class Payload
{
[JsonProperty("tag")]
public Guid TagID { get; set; }
[JsonProperty("probability")]
public double Probability { get; set; }
}
public async Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
public Task HandleApplicationMessageReceivedAsync(MqttApplicationMessageReceivedEventArgs eventArgs)
{
var message = eventArgs.ApplicationMessage.ConvertPayloadToString();
_logger.LogDebug($"Recieved [{eventArgs.ClientId}] " +
$"Topic: {eventArgs.ApplicationMessage.Topic} | Payload: {message} | QoS: {eventArgs.ApplicationMessage.QualityOfServiceLevel} | Retain: {eventArgs.ApplicationMessage.Retain}");
try
{
var payload = JsonConvert.DeserializeObject<Payload>(message);
var inputResponse = await _inputService.GetInputAsync(payload.TagID);
lock (_messageLock)
{
_messages.Add(new Message(inputResponse.Message.Device_id, inputResponse.Message.Date.UtcDateTime, payload.Probability));
}
}
catch (Exception ex)
{
_logger.LogError(ex, $"Could not handle application message.");
}
return ProcessJsonMessageAsync(message);
}
public async Task HandleConnectedAsync(MqttClientConnectedEventArgs eventArgs)
@ -107,7 +56,7 @@ namespace Birdmap.BLL.Services.CommunicationServices.Mqtt
_logger.LogInformation($"Connected. Auth result: {eventArgs.AuthenticateResult}. Subscribing to topic: {topic}");
await _mqttClient.SubscribeAsync(topic);
_hubTimer.Start();
StartMessageTimer();
}
catch (Exception ex)
{
@ -123,7 +72,7 @@ namespace Birdmap.BLL.Services.CommunicationServices.Mqtt
try
{
_hubTimer.Stop();
StopMessageTimer();
await _mqttClient.ConnectAsync(_options, CancellationToken.None);
}
catch (Exception ex)

14
Birdmap.BLL/Services/CommunationServices/Mqtt/MqttClientServiceProvider.cs
View File

@ -1,14 +0,0 @@
using Birdmap.BLL.Interfaces;
namespace Birdmap.BLL.Services.CommunicationServices.Mqtt
{
public class MqttClientServiceProvider
{
public IMqttClientService MqttClientService { get; }
public MqttClientServiceProvider(IMqttClientService mqttClientService)
{
MqttClientService = mqttClientService;
}
}
}

72
Birdmap.BLL/Services/CommunationServices/RabbitMq/RabbitMqClientService.cs Normal file
View File

@ -0,0 +1,72 @@
using Birdmap.BLL.Interfaces;
using Birdmap.BLL.Options;
using Birdmap.BLL.Services.CommunicationServices.Hubs;
using Microsoft.AspNetCore.SignalR;
using Microsoft.Extensions.Logging;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using System.Threading.Tasks;
namespace Birdmap.BLL.Services.CommunationServices.RabbitMq
{
internal class RabbitMqClientService : CommunicationServiceBase, IDisposable
{
private readonly IConnection _connection;
private readonly IModel _channel;
public override bool IsConnected => throw new NotImplementedException();
public RabbitMqClientService(RabbitMqClientOptions options, ILogger<RabbitMqClientService> logger, IInputService inputService, IHubContext<DevicesHub, IDevicesHubClient> hubContext)
: base(logger, inputService, hubContext)
{
var factory = new ConnectionFactory()
{
HostName = options.Hostname,
Port = options.Port,
UserName = options.Username,
Password = options.Password,
AutomaticRecoveryEnabled = true,
};
_connection = factory.CreateConnection();
_channel = _connection.CreateModel();
_channel.ExchangeDeclare(exchange: "topic_logs", type: "topic");
var queueName = _channel.QueueDeclare().QueueName;
_channel.QueueBind(queue: queueName,
exchange: "topic_logs",
routingKey: options.Topic);
var consumer = new AsyncEventingBasicConsumer(_channel);
consumer.Received += OnRecieved;
_channel.BasicConsume(queue: queueName,
autoAck: true,
consumer: consumer);
StartMessageTimer();
}
private Task OnRecieved(object sender, BasicDeliverEventArgs eventArgs)
{
var props = eventArgs.BasicProperties;
var body = Encoding.UTF8.GetString(eventArgs.Body.ToArray());
_logger.LogDebug($"Recieved [{props.UserId}] " +
$"ConsumerTag: {eventArgs.ConsumerTag} | DeliveryTag: {eventArgs.DeliveryTag} | Payload: {body} | Priority: {props.Priority}");
return ProcessJsonMessageAsync(body);
}
public void Dispose()
{
StopMessageTimer();
_channel.Close();
_connection.Close();
}
}
}