| /trunk/WingMan/Communication/MqttCommunication.cs |
|---|
| @@ -0,0 +1,391 @@ |
|---|
| using System; |
| using System.Net; |
| using System.Text; |
| using System.Threading; |
| using System.Threading.Tasks; |
| using MQTTnet; |
| using MQTTnet.Client; |
| using MQTTnet.Extensions.ManagedClient; |
| using MQTTnet.Protocol; |
| using MQTTnet.Server; |
| using WingMan.Utilities; |
| using MqttClientConnectedEventArgs = MQTTnet.Client.MqttClientConnectedEventArgs; |
| using MqttClientDisconnectedEventArgs = MQTTnet.Client.MqttClientDisconnectedEventArgs; |
| |
| namespace WingMan.Communication |
| { |
| public class MqttCommunication : IDisposable |
| { |
| public delegate void ClientAuthenticationFailed(object sender, MqttAuthenticationFailureEventArgs e); |
| |
| 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 ServerAuthenticationFailed(object sender, MqttAuthenticationFailureEventArgs 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 string Password { get; set; } |
| |
| private CancellationToken CancellationToken { get; } |
| |
| public MqttCommunicationType Type { get; set; } |
| |
| public async void Dispose() |
| { |
| await Stop(); |
| } |
| |
| public event ClientAuthenticationFailed OnClientAuthenticationFailed; |
| |
| public event ServerAuthenticationFailed OnServerAuthenticationFailed; |
| |
| 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<bool> Start(MqttCommunicationType type, IPAddress ipAddress, int port, string nick, |
| string password) |
| { |
| Type = type; |
| IpAddress = ipAddress; |
| Port = port; |
| Nick = nick; |
| Password = password; |
| |
| switch (type) |
| { |
| case MqttCommunicationType.Client: |
| return await StartClient(); |
| case MqttCommunicationType.Server: |
| return await StartServer(); |
| } |
| |
| return false; |
| } |
| |
| private async Task<bool> StartClient() |
| { |
| var clientOptions = new MqttClientOptionsBuilder() |
| .WithTcpServer(IpAddress.ToString(), Port); |
| |
| // Setup and start a managed MQTT client. |
| var options = new ManagedMqttClientOptionsBuilder() |
| .WithClientOptions(clientOptions.Build()) |
| .Build(); |
| |
| BindClientHandlers(); |
| |
| await Client.SubscribeAsync( |
| new TopicFilterBuilder() |
| .WithTopic("lobby") |
| .Build() |
| ); |
| |
| await Client.SubscribeAsync( |
| new TopicFilterBuilder() |
| .WithTopic("exchange") |
| .Build() |
| ); |
| |
| await Client.SubscribeAsync( |
| new TopicFilterBuilder() |
| .WithTopic("execute") |
| .Build() |
| ); |
| |
| await Client.StartAsync(options); |
| |
| Running = true; |
| |
| return Running; |
| } |
| |
| private async Task StopClient() |
| { |
| UnbindClientHandlers(); |
| |
| await Client.StopAsync(); |
| } |
| |
| 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) |
| { |
| try |
| { |
| var load = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); |
| |
| e.ApplicationMessage.Payload = AES.Decrypt(e.ApplicationMessage.Payload, Password); |
| |
| var load2 = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); |
| |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| catch (Exception ex) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith( |
| _ => OnClientAuthenticationFailed?.Invoke(sender, new MqttAuthenticationFailureEventArgs(e, ex)), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| } |
| |
| private async void ClientOnConnectingFailed(object sender, MqttManagedProcessFailedEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnClientConnectionFailed?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async void ClientOnDisconnected(object sender, MqttClientDisconnectedEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnClientDisconnected?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async void ClientOnConnected(object sender, MqttClientConnectedEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnClientConnected?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async Task<bool> StartServer() |
| { |
| var optionsBuilder = new MqttServerOptionsBuilder() |
| .WithDefaultEndpointBoundIPAddress(IpAddress) |
| .WithSubscriptionInterceptor(MqttSubscriptionIntercept) |
| .WithConnectionValidator(MqttConnectionValidator) |
| .WithDefaultEndpointPort(Port); |
| |
| BindServerHandlers(); |
| |
| try |
| { |
| await Server.StartAsync(optionsBuilder.Build()); |
| |
| Running = true; |
| } |
| catch (Exception) |
| { |
| Running = false; |
| } |
| |
| return Running; |
| } |
| |
| private void MqttConnectionValidator(MqttConnectionValidatorContext context) |
| { |
| context.ReturnCode = MqttConnectReturnCode.ConnectionAccepted; |
| } |
| |
| private async Task StopServer() |
| { |
| UnbindServerHandlers(); |
| |
| await Server.StopAsync(); |
| } |
| |
| private void MqttSubscriptionIntercept(MqttSubscriptionInterceptorContext context) |
| { |
| if (context.TopicFilter.Topic != "lobby" && |
| context.TopicFilter.Topic != "exchange" && |
| context.TopicFilter.Topic != "execute") |
| { |
| 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) |
| { |
| try |
| { |
| var load = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); |
| |
| e.ApplicationMessage.Payload = AES.Decrypt(e.ApplicationMessage.Payload, Password); |
| |
| var load2 = Encoding.UTF8.GetString(e.ApplicationMessage.Payload); |
| |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnMessageReceived?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| catch (Exception ex) |
| { |
| foreach (var clientSessionStatus in await Server.GetClientSessionsStatusAsync()) |
| { |
| if (!string.Equals(clientSessionStatus.ClientId, e.ClientId, StringComparison.Ordinal)) |
| continue; |
| |
| await clientSessionStatus.DisconnectAsync(); |
| } |
| |
| await Task.Delay(0, CancellationToken).ContinueWith( |
| _ => OnServerAuthenticationFailed?.Invoke(sender, new MqttAuthenticationFailureEventArgs(e, ex)), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| } |
| |
| private async void ServerOnClientUnsubscribedTopic(object sender, MqttClientUnsubscribedTopicEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnClientUnsubscribed?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async void ServerOnClientSubscribedTopic(object sender, MqttClientSubscribedTopicEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnClientSubscribed?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async void ServerOnClientDisconnected(object sender, MQTTnet.Server.MqttClientDisconnectedEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnServerClientDisconnected?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| private async void ServerOnClientConnected(object sender, MQTTnet.Server.MqttClientConnectedEventArgs e) |
| { |
| await Task.Delay(0, CancellationToken).ContinueWith(_ => OnServerClientConnected?.Invoke(sender, e), |
| CancellationToken, TaskContinuationOptions.None, TaskScheduler); |
| } |
| |
| 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(); |
| break; |
| case MqttCommunicationType.Client: |
| await StopClient(); |
| break; |
| } |
| |
| Running = false; |
| } |
| |
| public async Task Broadcast(string topic, byte[] payload) |
| { |
| var encryptedPayload = AES.Encrypt(payload, Password); |
| |
| var load = Encoding.UTF8.GetString(encryptedPayload); |
| |
| switch (Type) |
| { |
| case MqttCommunicationType.Client: |
| await Client.PublishAsync(new ManagedMqttApplicationMessage |
| { |
| ApplicationMessage = new MqttApplicationMessage {Topic = topic, Payload = encryptedPayload } |
| }); |
| break; |
| case MqttCommunicationType.Server: |
| await Server.PublishAsync(new MqttApplicationMessage {Topic = topic, Payload = encryptedPayload }); |
| break; |
| } |
| } |
| } |
| } |
