320 lines
12 KiB
C#
320 lines
12 KiB
C#
using System.IO;
|
|
using System;
|
|
using System.Collections.Generic;
|
|
using System.Text;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using CommonUsage.Protocols.VDA5050.Messages;
|
|
using MQTTnet;
|
|
using MQTTnet.Client;
|
|
using MQTTnet.Extensions.ManagedClient;
|
|
using MQTTnet.Packets;
|
|
using MQTTnet.Protocol;
|
|
using MQTTnet.Server;
|
|
using Newtonsoft.Json;
|
|
using FundamentalLib;
|
|
using CommonUsage.Protocols.VDA5050.Objects;
|
|
using FundamentalLib.MiscHelpers;
|
|
|
|
namespace CommonUsage.Protocols.VDA5050
|
|
{
|
|
public class MQTTCommunication : IVDACommunicationProtocol
|
|
{
|
|
private readonly string _host;
|
|
private readonly int _port;
|
|
private readonly string _orderTopic = "vda5050/frldAGV/order";
|
|
private readonly string _connectionTopic = "vda5050/frldAGV/connection";
|
|
private readonly string _instanceAction = "vda5050/frldAGV/instantActions";
|
|
private readonly string _factsheet = "vda5050/frldAGV/factsheet";
|
|
private readonly string _changeCarFields = "vda5050/frldAGV/changeCarFields";
|
|
|
|
private IManagedMqttClient _client;
|
|
private IManagedMqttClient _visualizationClient;
|
|
|
|
public MQTTCommunication(string host, int port)
|
|
{
|
|
_host = host;
|
|
_port = port;
|
|
InitializeClient();
|
|
InitializeVisualizationClient();
|
|
}
|
|
|
|
private void InitializeClient()
|
|
{
|
|
var willMessage = new connectionMessage()
|
|
{
|
|
headerId = 1,
|
|
timestamp = DateTime.Now,
|
|
version = "00",
|
|
manufacturer = "frld",
|
|
serialNumber = "test-01",
|
|
connectionState = "CONNECTIONBROKEN"
|
|
};
|
|
|
|
var mqttClientOptions = new MqttClientOptionsBuilder()
|
|
.WithClientId("AGV-Client-frldAGV")
|
|
.WithTcpServer(_host, _port)
|
|
.WithWillTopic(_connectionTopic)
|
|
.WithWillPayload(JsonConvert.SerializeObject(willMessage))
|
|
.WithWillRetain(true)
|
|
.Build();
|
|
|
|
var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
|
|
.WithClientOptions(mqttClientOptions)
|
|
.WithMaxPendingMessages(20)
|
|
.WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
|
|
.Build();
|
|
|
|
_client = new MqttFactory().CreateManagedMqttClient();
|
|
_client.StartAsync(managedMqttClientOptions).GetAwaiter().GetResult();
|
|
Console.WriteLine($" >> MQTT client initialized and connected to broker at {_host} - {_port}");
|
|
}
|
|
|
|
private void InitializeVisualizationClient()
|
|
{
|
|
var mqttClientOptions = new MqttClientOptionsBuilder()
|
|
.WithClientId("AGV-Visualization")
|
|
.WithTcpServer(_host, _port)
|
|
.Build();
|
|
|
|
var managedMqttClientOptions = new ManagedMqttClientOptionsBuilder()
|
|
.WithClientOptions(mqttClientOptions)
|
|
.WithMaxPendingMessages(10) // Prevent overloading
|
|
.WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
|
|
.Build();
|
|
|
|
_visualizationClient = new MqttFactory().CreateManagedMqttClient();
|
|
_visualizationClient.StartAsync(managedMqttClientOptions).GetAwaiter().GetResult();
|
|
Console.WriteLine("MQTT Visualization client initialized.");
|
|
}
|
|
|
|
public void SetUpImmediateCommandListener(Action<string> onChangeCarFieldsReceived)
|
|
{
|
|
Console.WriteLine($"Subscribed to the topic: {_changeCarFields}");
|
|
_client.SubscribeAsync(_changeCarFields).GetAwaiter().GetResult();
|
|
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if (e.ApplicationMessage.Topic == _changeCarFields)
|
|
{
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
LogMessage($"RECEIVE-{e.ApplicationMessage.Topic}", e.ApplicationMessage.Topic, payload);
|
|
//var script = JsonConvert.DeserializeObject<string>(payload);
|
|
Console.WriteLine($"Change car field: {payload}");
|
|
onChangeCarFieldsReceived(payload);
|
|
}
|
|
};
|
|
}
|
|
|
|
|
|
public void SetUpInstanActionListener(Action<instanceAction> onInstanceActionReceived)
|
|
{
|
|
Console.WriteLine($"Subscribed to the topic: {_instanceAction}");
|
|
|
|
// Subscribe to the instanceAction topic
|
|
_client.SubscribeAsync(_instanceAction).GetAwaiter().GetResult();
|
|
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if (e.ApplicationMessage.Topic == _instanceAction)
|
|
{
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
LogMessage($"RECEIVE-{e.ApplicationMessage.Topic}", e.ApplicationMessage.Topic, payload);
|
|
var actions = JsonConvert.DeserializeObject<instanceAction>(payload);
|
|
Console.WriteLine($"Received instance action: Header ID = {actions.headerId}, Timestamp = {actions.timestamp}");
|
|
foreach (var action in actions.actions)
|
|
{
|
|
Console.WriteLine($"Action ID: {action.actionId}, Type: {action.actionType}");
|
|
}
|
|
|
|
onInstanceActionReceived(actions);
|
|
}
|
|
};
|
|
}
|
|
|
|
public async Task PublishConnectionStatus(string status)
|
|
{
|
|
var message = new connectionMessage()
|
|
{
|
|
headerId = 1,
|
|
timestamp = DateTime.Now,
|
|
version = "00",
|
|
manufacturer = "frld",
|
|
serialNumber = "test-01",
|
|
connectionState = status
|
|
};
|
|
|
|
var payload = JsonConvert.SerializeObject(message);
|
|
|
|
var content = new MqttApplicationMessageBuilder()
|
|
.WithTopic(_connectionTopic)
|
|
.WithPayload(payload)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)
|
|
.WithRetainFlag(true)
|
|
.Build();
|
|
|
|
await _client.EnqueueAsync(content);
|
|
LogMessage($"SEND-{_connectionTopic}", _connectionTopic, payload);
|
|
|
|
//await SendMessageAsync(message, _connectionTopic);
|
|
}
|
|
|
|
public async Task PublishFactSheet(factsheetMessage message)
|
|
{
|
|
|
|
//await SendMessageAsync(message, _factsheet);
|
|
var payload = JsonConvert.SerializeObject(message);
|
|
|
|
var content = new MqttApplicationMessageBuilder()
|
|
.WithTopic(_factsheet)
|
|
.WithPayload(payload)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
|
|
//.WithRetainFlag(true)
|
|
.Build();
|
|
|
|
await _client.EnqueueAsync(content);
|
|
LogMessage($"SEND-{_factsheet}", _factsheet, payload);
|
|
|
|
}
|
|
|
|
public void SetupOrderListener(Action<orderMessage> orderReceived)
|
|
{
|
|
// Subscribe to the orders topic
|
|
_client.SubscribeAsync(_orderTopic, MqttQualityOfServiceLevel.AtMostOnce).GetAwaiter().GetResult();
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if (e.ApplicationMessage.Topic == _orderTopic)
|
|
{
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
LogMessage($"RECEIVE-{e.ApplicationMessage.Topic}", e.ApplicationMessage.Topic, payload);
|
|
var order = JsonConvert.DeserializeObject<orderMessage>(payload);
|
|
|
|
// Save the order message to a file for debugging
|
|
// SaveOrderToFile(payload);
|
|
orderReceived(order);
|
|
}
|
|
await Task.CompletedTask;
|
|
};
|
|
}
|
|
|
|
private void SaveOrderToFile(string orderJson)
|
|
{
|
|
try
|
|
{
|
|
// Specify the file path (e.g., orders_log.txt in the current directory)
|
|
string filePath = "orders_log.txt";
|
|
|
|
// Append the order JSON along with a timestamp
|
|
File.AppendAllText(filePath, $"{DateTime.UtcNow:yyyy-MM-dd HH:mm:ss} - {orderJson}{Environment.NewLine}");
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// Handle any exceptions that occur while writing to the file
|
|
Console.WriteLine($"Failed to save order to file: {ex.Message}");
|
|
}
|
|
}
|
|
|
|
public async Task SendMessageAsync<T>(T message, string topic)
|
|
{
|
|
|
|
var payload = JsonConvert.SerializeObject(message);
|
|
|
|
var content = new MqttApplicationMessageBuilder()
|
|
.WithTopic(topic)
|
|
.WithPayload(payload)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
|
|
.Build();
|
|
|
|
await _client.EnqueueAsync(content);
|
|
|
|
if (topic != "vda5050/frldAGV/visualization")
|
|
{
|
|
|
|
LogMessage($"SEND-{topic}", topic, payload);
|
|
}
|
|
|
|
}
|
|
|
|
public async Task SendVisualizationMessageAsync<T>(T message, string topic)
|
|
{
|
|
if (_visualizationClient == null) return; // Ensure client is initialized
|
|
|
|
var payload = JsonConvert.SerializeObject(message);
|
|
|
|
var content = new MqttApplicationMessageBuilder()
|
|
.WithTopic(topic)
|
|
.WithPayload(payload)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce) // QoS 0 for lightweight visualization
|
|
.Build();
|
|
|
|
if (_visualizationClient.PendingApplicationMessagesCount < 5) // Prevent flooding
|
|
{
|
|
await _visualizationClient.EnqueueAsync(content);
|
|
}
|
|
else
|
|
{
|
|
Console.WriteLine("Skipping visualization update to avoid MQTT congestion.");
|
|
}
|
|
}
|
|
|
|
|
|
public void LogMessage(string direction, string topic, string payload)
|
|
{
|
|
string formattedPayload = payload;
|
|
string logDirectory = "Logs"; // Directory for log files
|
|
var filePreName = direction;
|
|
filePreName = filePreName.Replace("/", "_");
|
|
string logFilePath = Path.Combine(logDirectory, filePreName + $"-{DateTime.Now:yyyy-MM-dd}.log");
|
|
|
|
if(!Directory.Exists(logDirectory)) Directory.CreateDirectory(logDirectory);
|
|
// Try to parse the payload as JSON and pretty-print it
|
|
try
|
|
{
|
|
var jsonObject = JsonConvert.DeserializeObject(payload);
|
|
formattedPayload = JsonConvert.SerializeObject(jsonObject, Formatting.Indented);
|
|
}
|
|
catch (JsonReaderException)
|
|
{
|
|
// If the payload is not valid JSON, just leave it as is
|
|
formattedPayload = payload;
|
|
}
|
|
|
|
string logMessage = $"[{DateTime.Now:HH:mm:ss}] [{direction}] Topic: {topic}, Payload:\n{formattedPayload}\n";
|
|
RotateLogFile(logFilePath, logDirectory);
|
|
|
|
try
|
|
{
|
|
File.AppendAllText(logFilePath, logMessage + Environment.NewLine);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Console.WriteLine($"Error writing to log file: {ex.Message}");
|
|
}
|
|
// Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] [{direction}] Topic: {topic}, Payload: {formattedPayload}");
|
|
// DLog.Log($"[{DateTime.Now:HH:mm:ss}] [{direction}] Topic: {topic}, Payload: {formattedPayload}");
|
|
}
|
|
|
|
|
|
private void RotateLogFile(string logFilePath, string logDirectory)
|
|
{
|
|
const long maxFileSize = 10 * 1024 * 1024; // 10 MB in bytes
|
|
|
|
FileInfo fileInfo = new FileInfo(logFilePath);
|
|
if (fileInfo.Exists && fileInfo.Length > maxFileSize)
|
|
{
|
|
string archivePath = Path.Combine(logDirectory, $"log_{DateTime.Now:yyyy-MM-dd_HH-mm-ss}.log");
|
|
|
|
try
|
|
{
|
|
File.Move(logFilePath, archivePath); // Rename the current log file
|
|
Console.WriteLine($"Log file rotated: {archivePath}");
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Console.WriteLine($"Error rotating log file: {ex.Message}");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|