Proof Of Concept initial
Adds initial files for a proof-of-concept project, including Dockerfile, .gitignore, project files, and MQTT-related services. This commit sets up the basic structure and configuration for exploring and validating the core concepts of the project.
This commit is contained in:
@@ -1,20 +1,61 @@
|
||||
using Microsoft.Extensions.Options;
|
||||
using MQTTnet;
|
||||
|
||||
namespace ProofOfConcept.Services;
|
||||
|
||||
public class MQTTClient
|
||||
public class MQTTClient : IHostedService
|
||||
{
|
||||
private ILogger<MQTTClient> logger;
|
||||
private MQTTClientConfiguration configuration;
|
||||
private MQTTServerConfiguration serverConfiguration;
|
||||
|
||||
public MQTTClient(ILogger<MQTTClient> logger, IOptionsMonitor<MQTTClientConfiguration> options)
|
||||
private readonly IMqttClient client;
|
||||
|
||||
public MQTTClient(ILogger<MQTTClient> logger, IOptions<MQTTClientConfiguration> options, IOptions<MQTTServerConfiguration> serverOptions, IMessageProcessor messageProcessor)
|
||||
{
|
||||
this.logger = logger;
|
||||
this.configuration = options.CurrentValue;
|
||||
|
||||
options.OnChange(newValue =>
|
||||
this.configuration = options.Value;
|
||||
this.serverConfiguration = serverOptions.Value;
|
||||
|
||||
client = new MqttClientFactory().CreateMqttClient();
|
||||
|
||||
this.client.ApplicationMessageReceivedAsync += (e) =>
|
||||
{
|
||||
this.configuration = newValue;
|
||||
logger.LogInformation("Configuration of {ClassName} changed", nameof(MQTTClient));
|
||||
});
|
||||
logger.LogInformation("Message received: {Message}", e.ApplicationMessage.Payload);
|
||||
messageProcessor.ProcessMessage(e.ApplicationMessage.Payload.ToString());
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
this.logger.LogTrace("Stating...");
|
||||
|
||||
MqttClientOptions options = new MqttClientOptionsBuilder()
|
||||
.WithTcpServer("localhost", this.serverConfiguration.Port)
|
||||
.WithClientId(this.serverConfiguration.LocalClient.ClientID)
|
||||
.WithCredentials(this.serverConfiguration.LocalClient.Username, this.serverConfiguration.LocalClient.Password)
|
||||
.Build();
|
||||
await this.client.ConnectAsync(options, cancellationToken);
|
||||
this.logger.LogTrace("Connected");
|
||||
|
||||
await this.client.SubscribeAsync("telemetry", cancellationToken: cancellationToken);
|
||||
this.logger.LogTrace("Subscribed");
|
||||
|
||||
this.logger.LogInformation("Started");
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
logger.LogTrace("Stopping...");
|
||||
|
||||
await this.client.UnsubscribeAsync("telemetry", cancellationToken: cancellationToken);
|
||||
this.logger.LogTrace("Unsubscribed");
|
||||
|
||||
await this.client.DisconnectAsync(cancellationToken: cancellationToken);
|
||||
this.logger.LogTrace("Disconnected");
|
||||
|
||||
logger.LogInformation("Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,24 +1,154 @@
|
||||
using System.Net;
|
||||
using Microsoft.Extensions.Options;
|
||||
using MQTTnet.Protocol;
|
||||
using MQTTnet.Server;
|
||||
|
||||
namespace ProofOfConcept.Services;
|
||||
|
||||
public class MQTTServer
|
||||
public class MQTTServer : IHostedService
|
||||
{
|
||||
private ILogger<MQTTServer> logger;
|
||||
private MQTTServerConfiguration configuration;
|
||||
private readonly ILogger<MQTTServer> logger;
|
||||
private readonly MQTTServerConfiguration configuration;
|
||||
|
||||
public MQTTServer(ILogger<MQTTServer> logger, IOptionsMonitor<MQTTServerConfiguration> options)
|
||||
private readonly MqttServer server;
|
||||
|
||||
public MQTTServer(ILogger<MQTTServer> logger, IOptions<MQTTServerConfiguration> options)
|
||||
{
|
||||
this.logger = logger;
|
||||
this.configuration = options.CurrentValue;
|
||||
this.configuration = options.Value;
|
||||
|
||||
options.OnChange(newValue =>
|
||||
MqttServerOptions mqttServerOptions = new MqttServerOptionsBuilder().WithDefaultEndpoint().WithDefaultEndpointPort(this.configuration.Port).WithKeepAlive().Build();
|
||||
this.server = new MqttServerFactory().CreateMqttServer(mqttServerOptions);
|
||||
|
||||
this.server.ClientConnectedAsync += (e) =>
|
||||
{
|
||||
this.configuration = newValue;
|
||||
logger.LogInformation("Configuration of {ClassName} changed", nameof(MQTTServer));
|
||||
});
|
||||
logger.LogInformation("Client connected from: {IPAddress}. ClientID: {ClientID}", e.RemoteEndPoint.ToString(), e.ClientId);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
this.server.ClientDisconnectedAsync += (e) =>
|
||||
{
|
||||
logger.LogInformation("Client {ClientID} disconnected from: {IPAddress}. Reason: {ReasonString} ({ReasonCode})", e.ClientId, e.RemoteEndPoint.ToString(), e.ReasonString, e.ReasonCode);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
this.server.ClientSubscribedTopicAsync += (e) =>
|
||||
{
|
||||
logger.LogInformation("Client {ClientID} subscribed to topic: {Topic}", e.ClientId, e.TopicFilter.Topic);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
this.server.ClientUnsubscribedTopicAsync += (e) =>
|
||||
{
|
||||
logger.LogInformation("Client {ClientID} unsubscribed from topic: {Topic}", e.ClientId, e.TopicFilter);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
this.server.InterceptingPublishAsync += (e) =>
|
||||
{
|
||||
if (e.ApplicationMessage.Topic == "telemetry" && e.ClientId != this.configuration.TelemetryClient.ClientID)
|
||||
{
|
||||
logger.LogWarning("Client {ClientID} tried to publish to telemetry topic", e.ClientId);
|
||||
|
||||
e.Response.ReasonCode = MqttPubAckReasonCode.NotAuthorized;
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
logger.LogInformation("Client {ClientID} published message to topic: {Topic}", e.ClientId, e.ApplicationMessage.Topic);
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
|
||||
this.server.ValidatingConnectionAsync += (e) =>
|
||||
{
|
||||
//Local client (by clientID)
|
||||
if (e.ClientId == this.configuration.LocalClient.ClientID)
|
||||
{
|
||||
//Not local connection (IP is not loopback)
|
||||
if (e.RemoteEndPoint is IPEndPoint ipEndPoint && !IPAddress.IsLoopback(ipEndPoint.Address))
|
||||
{
|
||||
logger.LogWarning("Local client tried to connect from non-loopback address: {IPAddress}", ipEndPoint.ToString());
|
||||
e.ReasonCode = MqttConnectReasonCode.NotAuthorized;
|
||||
}
|
||||
|
||||
//Invalid username or password
|
||||
else if (e.UserName != configuration.LocalClient.Username || e.Password != configuration.LocalClient.Password)
|
||||
{
|
||||
logger.LogWarning("Local client tried to connect with invalid credentials");
|
||||
e.ReasonCode = MqttConnectReasonCode.NotAuthorized;
|
||||
}
|
||||
}
|
||||
|
||||
//Telemetry client
|
||||
else if (e.ClientId == this.configuration.TelemetryClient.ClientID)
|
||||
{
|
||||
if (e.UserName != configuration.TelemetryClient.Username || e.Password != configuration.TelemetryClient.Password)
|
||||
{
|
||||
logger.LogWarning("Telemetry client tried to connect with invalid credentials");
|
||||
e.ReasonCode = MqttConnectReasonCode.NotAuthorized;
|
||||
}
|
||||
}
|
||||
|
||||
//Observer
|
||||
else if ((this.configuration.Observer.ClientID is not null && e.ClientId == this.configuration.Observer.ClientID) || this.configuration.Observer.ClientID is null)
|
||||
{
|
||||
if (e.UserName != configuration.Observer.Username || e.Password != configuration.Observer.Password)
|
||||
{
|
||||
logger.LogWarning("Observer tried to connect with invalid credentials");
|
||||
e.ReasonCode = MqttConnectReasonCode.NotAuthorized;
|
||||
}
|
||||
}
|
||||
|
||||
else
|
||||
e.ReasonCode = MqttConnectReasonCode.Banned;
|
||||
|
||||
return Task.CompletedTask;
|
||||
};
|
||||
}
|
||||
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
this.logger.LogTrace("Starting MQTT Server...");
|
||||
|
||||
await this.server.StartAsync();
|
||||
|
||||
this.logger.LogInformation("MQTT Server started");
|
||||
}
|
||||
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
this.logger.LogTrace("Stopping MQTT Server...");
|
||||
|
||||
await this.server.StopAsync();
|
||||
|
||||
this.logger.LogInformation("MQTT Server stopped");
|
||||
}
|
||||
}
|
||||
|
||||
public class MQTTServerConfiguration
|
||||
{
|
||||
public ushort Port { get; set; } = 1883;
|
||||
public MQTTAuthenticationData TelemetryClient { get; set; } = new MQTTAuthenticationData("telemetry", "fleet", "9R0FXNs5NEIEbI4"); //Must have a known client id
|
||||
public MQTTAuthenticationData LocalClient { get; set; } = new MQTTAuthenticationData("local", "client", "xuhvoc-parrox-xuzsE4"); //Must have a known client id
|
||||
public MQTTAuthenticationData Observer { get; set; } = new MQTTAuthenticationData("observer", "JC740d4c4E9dveh");
|
||||
}
|
||||
|
||||
public class MQTTAuthenticationData
|
||||
{
|
||||
public string? ClientID { get; set; }
|
||||
public string Username { get; set; }
|
||||
public string Password { get; set; }
|
||||
|
||||
public MQTTAuthenticationData(string clientId, string username, string password)
|
||||
{
|
||||
ClientID = clientId;
|
||||
Username = username;
|
||||
Password = password;
|
||||
}
|
||||
|
||||
public MQTTAuthenticationData(string username, string password)
|
||||
{
|
||||
ClientID = null;
|
||||
Username = username;
|
||||
Password = password;
|
||||
}
|
||||
}
|
||||
@@ -1,20 +1,31 @@
|
||||
using Microsoft.Extensions.Caching.Memory;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace ProofOfConcept.Services;
|
||||
|
||||
public class MessageProcessor
|
||||
public interface IMessageProcessor
|
||||
{
|
||||
private ILogger<MessageProcessor> logger;
|
||||
Task ProcessMessage(string jsonMessage);
|
||||
}
|
||||
|
||||
public class MessageProcessor : IMessageProcessor
|
||||
{
|
||||
private readonly ILogger<MessageProcessor> logger;
|
||||
private MessageProcessorConfiguration configuration;
|
||||
|
||||
public MessageProcessor(ILogger<MessageProcessor> logger, IOptionsMonitor<MessageProcessorConfiguration> options)
|
||||
private readonly IMemoryCache memoryCache;
|
||||
|
||||
public MessageProcessor(ILogger<MessageProcessor> logger, IOptions<MessageProcessorConfiguration> options, IMemoryCache memoryCache)
|
||||
{
|
||||
this.logger = logger;
|
||||
this.configuration = options.CurrentValue;
|
||||
this.configuration = options.Value;
|
||||
|
||||
options.OnChange(newValue =>
|
||||
{
|
||||
this.configuration = newValue;
|
||||
logger.LogInformation("Configuration of {ClassName} changed", nameof(MessageProcessor));
|
||||
});
|
||||
this.memoryCache = memoryCache;
|
||||
}
|
||||
|
||||
public async Task ProcessMessage(string jsonMessage)
|
||||
{
|
||||
this.logger.LogTrace("Processing message from Tesla: {Message}", jsonMessage);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user