WingMan – Rev 7
?pathlinks?
using System;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Extensions.ManagedClient;
using MQTTnet.Server;
using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs;
using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs;
namespace WingMan.Communication
{
public class MQTTCommunication : IDisposable
{
public delegate void ClientConnected(object sender, MqttClientConnectedEventArgs e);
public delegate void ClientConnectionFailed(object sender, MqttManagedProcessFailedEventArgs e);
public delegate void ClientDisconnected(object sender, MqttClientDisconnectedEventArgs e);
public delegate void ClientSubscribed(object sender, MqttClientSubscribedTopicEventArgs e);
public delegate void ClientUnsubscribed(object sender, MqttClientUnsubscribedTopicEventArgs e);
public delegate void MessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e);
public delegate void ServerClientConnected(object sender, MQTTnet.Server.MqttClientConnectedEventArgs e);
public delegate void ServerClientDisconnected(object sender, MQTTnet.Server.MqttClientDisconnectedEventArgs e);
public delegate void ServerStarted(object sender, EventArgs e);
public delegate void ServerStopped(object sender, EventArgs e);
public MQTTCommunication(TaskScheduler taskScheduler, CancellationToken cancellationToken)
{
TaskScheduler = taskScheduler;
CancellationToken = cancellationToken;
Client = new MqttFactory().CreateManagedMqttClient();
Server = new MqttFactory().CreateMqttServer();
}
private TaskScheduler TaskScheduler { get; }
private IManagedMqttClient Client { get; }
private IMqttServer Server { get; }
public bool Running { get; set; }
public string Nick { get; set; }
private IPAddress IPAddress { get; set; }
private int Port { get; set; }
private CancellationToken CancellationToken { get; }
public MQTTCommunicationType Type { get; set; }
public async void Dispose()
{
await Stop();
}
public event MessageReceived OnMessageReceived;
public event ClientConnected OnClientConnected;
public event ClientDisconnected OnClientDisconnected;
public event ClientConnectionFailed OnClientConnectionFailed;
public event ClientUnsubscribed OnClientUnsubscribed;
public event ClientSubscribed OnClientSubscribed;
public event ServerClientDisconnected OnServerClientDisconnected;
public event ServerClientConnected OnServerClientConnected;
public event ServerStarted OnServerStarted;
public event ServerStopped OnServerStopped;
public async Task Start(MQTTCommunicationType type, IPAddress ipAddress, int port, string nick)
{
Type = type;
IPAddress = ipAddress;
Port = port;
Nick = nick;
switch (type)
{
case MQTTCommunicationType.Client:
await StartClient().ConfigureAwait(false);
break;
case MQTTCommunicationType.Server:
await StartServer().ConfigureAwait(false);
break;
}
}
private async Task StartClient()
{
var clientOptions = new MqttClientOptionsBuilder()
.WithTcpServer(IPAddress.ToString(), Port);
// Setup and start a managed MQTT client.
var options = new ManagedMqttClientOptionsBuilder()
.WithAutoReconnectDelay(TimeSpan.FromSeconds(5))
.WithClientOptions(clientOptions.Build())
.Build();
BindClientHandlers();
await Client.SubscribeAsync(
new TopicFilterBuilder()
.WithTopic("lobby")
.Build()
).ConfigureAwait(false);
await Client.SubscribeAsync(
new TopicFilterBuilder()
.WithTopic("exchange")
.Build()
).ConfigureAwait(false);
await Client.StartAsync(options).ConfigureAwait(false);
Running = true;
}
private async Task StopClient()
{
UnbindClientHandlers();
await Client.StopAsync().ConfigureAwait(false);
}
public void BindClientHandlers()
{
Client.Connected += ClientOnConnected;
Client.Disconnected += ClientOnDisconnected;
Client.ConnectingFailed += ClientOnConnectingFailed;
Client.ApplicationMessageReceived += ClientOnApplicationMessageReceived;
}
public void UnbindClientHandlers()
{
Client.Connected -= ClientOnConnected;
Client.Disconnected -= ClientOnDisconnected;
Client.ConnectingFailed -= ClientOnConnectingFailed;
Client.ApplicationMessageReceived -= ClientOnApplicationMessageReceived;
}
private async void ClientOnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler);
}
private async void ClientOnConnectingFailed(object sender, MqttManagedProcessFailedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnClientConnectionFailed?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ClientOnDisconnected(object sender, MqttClientDisconnectedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnClientDisconnected?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ClientOnConnected(object sender, MqttClientConnectedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnClientConnected?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async Task StartServer()
{
var optionsBuilder = new MqttServerOptionsBuilder()
.WithDefaultEndpointBoundIPAddress(IPAddress)
.WithSubscriptionInterceptor(MQTTSubscriptionIntercept)
.WithDefaultEndpointPort(Port);
BindServerHandlers();
await Server.StartAsync(optionsBuilder.Build()).ConfigureAwait(false);
Running = true;
}
private async Task StopServer()
{
UnbindServerHandlers();
await Server.StopAsync().ConfigureAwait(false);
}
private void MQTTSubscriptionIntercept(MqttSubscriptionInterceptorContext context)
{
if (context.TopicFilter.Topic != "lobby" &&
context.TopicFilter.Topic != "exchange")
{
context.AcceptSubscription = false;
context.CloseConnection = true;
return;
}
context.AcceptSubscription = true;
context.CloseConnection = false;
}
private void BindServerHandlers()
{
Server.Started += ServerOnStarted;
Server.Stopped += ServerOnStopped;
Server.ClientConnected += ServerOnClientConnected;
Server.ClientDisconnected += ServerOnClientDisconnected;
Server.ClientSubscribedTopic += ServerOnClientSubscribedTopic;
Server.ClientUnsubscribedTopic += ServerOnClientUnsubscribedTopic;
Server.ApplicationMessageReceived += ServerOnApplicationMessageReceived;
}
private void UnbindServerHandlers()
{
Server.Started -= ServerOnStarted;
Server.Stopped -= ServerOnStopped;
Server.ClientConnected -= ServerOnClientConnected;
Server.ClientDisconnected -= ServerOnClientDisconnected;
Server.ClientSubscribedTopic -= ServerOnClientSubscribedTopic;
Server.ClientUnsubscribedTopic -= ServerOnClientUnsubscribedTopic;
Server.ApplicationMessageReceived -= ServerOnApplicationMessageReceived;
}
private async void ServerOnApplicationMessageReceived(object sender, MqttApplicationMessageReceivedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ServerOnClientUnsubscribedTopic(object sender, MqttClientUnsubscribedTopicEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnClientUnsubscribed?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ServerOnClientSubscribedTopic(object sender, MqttClientSubscribedTopicEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnClientSubscribed?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ServerOnClientDisconnected(object sender, MQTTnet.Server.MqttClientDisconnectedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnServerClientDisconnected?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private async void ServerOnClientConnected(object sender, MQTTnet.Server.MqttClientConnectedEventArgs e)
{
await Task.Delay(0).ContinueWith(_ => OnServerClientConnected?.Invoke(sender, e),
CancellationToken, TaskContinuationOptions.None, TaskScheduler).ConfigureAwait(false);
}
private void ServerOnStopped(object sender, EventArgs e)
{
OnServerStopped?.Invoke(sender, e);
}
private void ServerOnStarted(object sender, EventArgs e)
{
OnServerStarted?.Invoke(sender, e);
}
public async Task Stop()
{
switch (Type)
{
case MQTTCommunicationType.Server:
await StopServer().ConfigureAwait(false);
break;
case MQTTCommunicationType.Client:
await StopClient().ConfigureAwait(false);
break;
}
Running = false;
}
public async Task Broadcast(string topic, byte[] payload)
{
switch (Type)
{
case MQTTCommunicationType.Client:
await Client.PublishAsync(new ManagedMqttApplicationMessage
{
ApplicationMessage = new MqttApplicationMessage {Topic = topic, Payload = payload}
}).ConfigureAwait(false);
break;
case MQTTCommunicationType.Server:
await Server.PublishAsync(new MqttApplicationMessage {Topic = topic, Payload = payload})
.ConfigureAwait(false);
break;
}
}
}
}
Generated by GNU Enscript 1.6.5.90.