init commit
This commit is contained in:
zhaowei.huang
@@ -0,0 +1,297 @@
|
||||
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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user