297 lines
12 KiB
C#
297 lines
12 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text;
|
|
using System.Threading.Tasks;
|
|
using CommonUsage.Protocols.VDA5050.Messages;
|
|
using MQTTnet;
|
|
using MQTTnet.Client;
|
|
using MQTTnet.Extensions.ManagedClient;
|
|
using MQTTnet.Protocol;
|
|
using MQTTnet.Server;
|
|
using Newtonsoft.Json;
|
|
using SimpleCore;
|
|
using SimpleCore.Library;
|
|
|
|
namespace StandardScene.CarTypes
|
|
{
|
|
public class MasterMQTTCommunication
|
|
{
|
|
private IManagedMqttClient _client;
|
|
private IManagedMqttClient _visualizationClient; // Separate client for visualization
|
|
|
|
private const string OrderTopic = "vda5050/frldAGV/order";
|
|
private const string StateTopic = "vda5050/frldAGV/state";
|
|
|
|
private const string VisualizationTopic = "vda5050/frldAGV/visualization";
|
|
private readonly string _connectionTopic = "vda5050/frldAGV/connection";
|
|
private readonly string _instantAction = "vda5050/frldAGV/instantActions";
|
|
private readonly string _factsheet = "vda5050/frldAGV/factsheet";
|
|
private readonly string _carFields = "vda5050/frldAGV/carFields";
|
|
|
|
|
|
|
|
//192.168.123.5
|
|
public MasterMQTTCommunication()
|
|
{
|
|
// Initialize MQTT client and connect to broker
|
|
var mqttFactory = new MqttFactory();
|
|
_client = mqttFactory.CreateManagedMqttClient();
|
|
|
|
var mqttClientOptions = new MqttClientOptionsBuilder()
|
|
.WithTcpServer("localhost", 1883) // Connect to the local broker
|
|
.WithCleanSession(true)
|
|
.WithCleanStart(true)
|
|
.Build();
|
|
|
|
var managedOptions = new ManagedMqttClientOptionsBuilder()
|
|
.WithClientOptions(mqttClientOptions)
|
|
.WithMaxPendingMessages(20)
|
|
.WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
|
|
.Build();
|
|
|
|
_client.StartAsync(managedOptions).GetAwaiter().GetResult();
|
|
Console.WriteLine(" >>> Master-Control MQTT client connected to broker.");
|
|
|
|
// Initialize separate client for visualization
|
|
_visualizationClient = mqttFactory.CreateManagedMqttClient();
|
|
var visualizationOptions = new MqttClientOptionsBuilder()
|
|
.WithTcpServer("localhost", 1883)
|
|
.WithClientId("Master-Visualization")
|
|
.Build();
|
|
|
|
var visualizationManagedOptions = new ManagedMqttClientOptionsBuilder()
|
|
.WithClientOptions(visualizationOptions)
|
|
.WithMaxPendingMessages(10) // Lower queue size for visualization
|
|
.WithPendingMessagesOverflowStrategy(MqttPendingMessagesOverflowStrategy.DropOldestQueuedMessage)
|
|
.Build();
|
|
|
|
_visualizationClient.StartAsync(visualizationManagedOptions).GetAwaiter().GetResult();
|
|
Console.WriteLine(" >>> Master-Control Visualization MQTT client initialized.");
|
|
|
|
}
|
|
|
|
public async Task RestartClient()
|
|
{
|
|
Console.WriteLine(" >>> Restarting MQTT client...");
|
|
|
|
var stopTask = _client.StopAsync();
|
|
if (await Task.WhenAny(stopTask, Task.Delay(5000)) == stopTask)
|
|
{
|
|
Console.WriteLine(" >>> MQTT client stopped successfully.");
|
|
}
|
|
else
|
|
{
|
|
Console.WriteLine(" >>> Timeout while stopping MQTT client.");
|
|
}
|
|
|
|
// Dispose and reinitialize the client
|
|
//_client.Dispose();
|
|
|
|
var mqttFactory = new MqttFactory();
|
|
var newClient = mqttFactory.CreateManagedMqttClient();
|
|
|
|
var mqttClientOptions = new MqttClientOptionsBuilder()
|
|
.WithTcpServer("localhost", 1883)
|
|
.Build();
|
|
|
|
var managedOptions = new ManagedMqttClientOptionsBuilder()
|
|
.WithClientOptions(mqttClientOptions)
|
|
.Build();
|
|
|
|
await newClient.StartAsync(managedOptions);
|
|
Console.WriteLine(" >>> New MQTT client started.");
|
|
}
|
|
|
|
|
|
|
|
public void SubsribeToConnectionTopic()
|
|
{
|
|
_client.SubscribeAsync(_connectionTopic, MqttQualityOfServiceLevel.AtLeastOnce).GetAwaiter().GetResult();
|
|
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if (e.ApplicationMessage.Topic == _connectionTopic)
|
|
{
|
|
Console.WriteLine($" >>>>>>> Subscribed to topic: {e.ApplicationMessage.Topic}");
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
//LogMessage("RECEIVED", e.ApplicationMessage.Topic, payload);
|
|
var connectionStatusMessage = JsonConvert.DeserializeObject<connectionMessage>(payload);
|
|
var car = (VDA5050Car)SimpleLib.GetAllCars()
|
|
.FirstOrDefault(cc => cc is VDA5050Car vdaCar);
|
|
if (car == null)
|
|
{
|
|
Diagnosis.Post($"no car of serialNumber {connectionStatusMessage.serialNumber}");
|
|
|
|
}
|
|
|
|
car.ConnectionStatus = connectionStatusMessage.connectionState;
|
|
Console.WriteLine($">>>> Connection status: {connectionStatusMessage.connectionState}");
|
|
}
|
|
};
|
|
}
|
|
|
|
public void SubscribeToCarFields()
|
|
{
|
|
_client.SubscribeAsync(_carFields,MqttQualityOfServiceLevel.AtLeastOnce).GetAwaiter().GetResult();
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if (e.ApplicationMessage.Topic == _carFields)
|
|
{
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
Console.WriteLine(payload);
|
|
//TODO:处理字符串，拿想要的值
|
|
}
|
|
};
|
|
}
|
|
|
|
//public void SubsribeToFactSheetTopic()
|
|
//{
|
|
// _client.SubscribeAsync(_factsheet, MqttQualityOfServiceLevel.AtMostOnce).GetAwaiter().GetResult();
|
|
|
|
// _client.ApplicationMessageReceivedAsync += async e =>
|
|
// {
|
|
// if (e.ApplicationMessage.Topic == _factsheet)
|
|
// {
|
|
// //Console.WriteLine($" >>>>>>> Subscribed to topic: {e.ApplicationMessage.Topic}");
|
|
// var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
// //LogMessage("RECEIVED FactSheet", e.ApplicationMessage.Topic, payload);
|
|
// // Console.WriteLine($">>>> Connection status: {connectionStatusMessage.connectionState}");
|
|
// }
|
|
// };
|
|
//}
|
|
|
|
public void SubscribeToVisualization()
|
|
{
|
|
_visualizationClient.SubscribeAsync(VisualizationTopic, MqttQualityOfServiceLevel.AtMostOnce).GetAwaiter().GetResult();
|
|
Console.WriteLine($" >>> Subscribed to {VisualizationTopic}");
|
|
|
|
_visualizationClient.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
if(_visualizationClient.PendingApplicationMessagesCount > 5)
|
|
{
|
|
Console.WriteLine($"[WARNING] Dropping old visualization message. Pending: {_visualizationClient.PendingApplicationMessagesCount}");
|
|
return; // Skip processing to catch up
|
|
}
|
|
if (e.ApplicationMessage.Topic == VisualizationTopic)
|
|
{
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
//Console.WriteLine($"[DEBUG] Received Visualization Message: {payload}");
|
|
|
|
var state = JsonConvert.DeserializeObject<visualizationMessage>(payload);
|
|
var car = (VDA5050Car)SimpleLib.GetAllCars()
|
|
.FirstOrDefault(cc => cc is VDA5050Car vdaCar);
|
|
|
|
if (car != null)
|
|
{
|
|
car.UpdatePosition(state);
|
|
}
|
|
}
|
|
await Task.CompletedTask;
|
|
};
|
|
}
|
|
|
|
public void SubscribeToState()
|
|
{
|
|
_client.SubscribeAsync(StateTopic, MqttQualityOfServiceLevel.AtMostOnce).GetAwaiter().GetResult();
|
|
Console.WriteLine($"Subscribed to topic: {StateTopic}");
|
|
_client.ApplicationMessageReceivedAsync += async e =>
|
|
{
|
|
Console.WriteLine("e.ApplicationMessage.Topic: " + e.ApplicationMessage.Topic + " StateTopic: " + StateTopic);
|
|
if (e.ApplicationMessage.Topic == StateTopic)
|
|
{
|
|
Console.WriteLine($" >>>>>>> Subscribed to topic: {e.ApplicationMessage.Topic}");
|
|
var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
//LogMessage("RECEIVED", e.ApplicationMessage.Topic, payload);
|
|
// Console.WriteLine($" >> Topic: [{e.ApplicationMessage.Topic}] -- {payload}");
|
|
var state = JsonConvert.DeserializeObject<stateMessage>(payload);
|
|
var car = (VDA5050Car)SimpleLib.GetAllCars()
|
|
.FirstOrDefault(cc => cc is VDA5050Car vdaCar);
|
|
if (car == null)
|
|
{
|
|
Diagnosis.Post($"no car of serialNumber {state.serialNumber}");
|
|
|
|
}
|
|
car.IsPaused = state.paused;
|
|
car.UpdateState(state);
|
|
//car.UpdatePosition(state);
|
|
//Console.WriteLine($">> Received message on topic '{StateTopic}': {payload}");
|
|
// onTopicReceived(state);
|
|
}
|
|
await Task.CompletedTask;
|
|
};
|
|
}
|
|
|
|
//public void SubscribeToState()
|
|
//{
|
|
// _client.SubscribeAsync(StateTopic, MqttQualityOfServiceLevel.AtMostOnce).GetAwaiter().GetResult();
|
|
// _client.ApplicationMessageReceivedAsync += async e =>
|
|
// {
|
|
// if (e.ApplicationMessage.Topic == StateTopic)
|
|
// {
|
|
// var payload = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
|
|
// var state = JsonConvert.DeserializeObject<stateMessage>(payload);
|
|
// var car = (VDA5050Car)SimpleLib.GetAllCars().FirstOrDefault(cc => cc is VDA5050Car);
|
|
// if (car == null)
|
|
// {
|
|
// Diagnosis.Post($"no car of serialNumber {state.serialNumber}");
|
|
// }
|
|
// else
|
|
// {
|
|
// // Instead of calling UpdateState/UpdatePosition directly,
|
|
// // simply store the latest state message.
|
|
// car.SetLatestState(state);
|
|
// }
|
|
// }
|
|
// await Task.CompletedTask;
|
|
// };
|
|
//}
|
|
|
|
|
|
|
|
public async Task PublishTo(string order)
|
|
{
|
|
var message = new MqttApplicationMessageBuilder()
|
|
.WithTopic(OrderTopic)
|
|
.WithPayload(order)
|
|
.WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)
|
|
.Build();
|
|
//Console.WriteLine($"[MQTT-Master] Pending Messages before sending: {_client.PendingApplicationMessagesCount}");
|
|
|
|
await _client.EnqueueAsync(message);
|
|
//Console.WriteLine($"[MQTT-Master] Sent order message. Pending Messages after sending: {_client.PendingApplicationMessagesCount}");
|
|
|
|
//LogMessage("Sent Order", OrderTopic, order);
|
|
}
|
|
|
|
public async Task PublishInstantActions(string actions)
|
|
{
|
|
var message = new MqttApplicationMessageBuilder()
|
|
.WithTopic(_instantAction)
|
|
.WithPayload(actions)
|
|
.Build();
|
|
|
|
await _client.EnqueueAsync(message);
|
|
|
|
}
|
|
|
|
public void LogMessage(string direction, string topic, string payload)
|
|
{
|
|
string formattedPayload = payload;
|
|
|
|
// 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;
|
|
}
|
|
|
|
Console.WriteLine($"[{DateTime.Now:HH:mm:ss}] [{direction}] Topic: {topic}, Payload: {formattedPayload}");
|
|
}
|
|
}
|
|
} |