@@ -4,11 +4,14 @@ namespace MQTTnet.Core.Client | |||||
{ | { | ||||
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs | public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs | ||||
{ | { | ||||
public MqttApplicationMessageReceivedEventArgs(MqttApplicationMessage applicationMessage) | |||||
public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage) | |||||
{ | { | ||||
ClientId = clientId; | |||||
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); | ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); | ||||
} | } | ||||
public string ClientId { get; } | |||||
public MqttApplicationMessage ApplicationMessage { get; } | public MqttApplicationMessage ApplicationMessage { get; } | ||||
} | } | ||||
} | } |
@@ -305,7 +305,7 @@ namespace MQTTnet.Core.Client | |||||
try | try | ||||
{ | { | ||||
var applicationMessage = publishPacket.ToApplicationMessage(); | var applicationMessage = publishPacket.ToApplicationMessage(); | ||||
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(applicationMessage)); | |||||
ApplicationMessageReceived?.Invoke(this, new MqttApplicationMessageReceivedEventArgs(_options.ClientId, applicationMessage)); | |||||
} | } | ||||
catch (Exception exception) | catch (Exception exception) | ||||
{ | { | ||||
@@ -2,13 +2,12 @@ | |||||
using System.Collections.Generic; | using System.Collections.Generic; | ||||
using System.Linq; | using System.Linq; | ||||
using System.Threading.Tasks; | using System.Threading.Tasks; | ||||
using MQTTnet.Core.Packets; | |||||
namespace MQTTnet.Core.Client | namespace MQTTnet.Core.Client | ||||
{ | { | ||||
public static class MqttClientExtensions | public static class MqttClientExtensions | ||||
{ | { | ||||
public static Task PublishAsync(this IMqttClient client, params MqttApplicationMessage[] applicationMessages) | |||||
public static Task PublishAsync(this IApplicationMessagePublisher client, params MqttApplicationMessage[] applicationMessages) | |||||
{ | { | ||||
if (client == null) throw new ArgumentNullException(nameof(client)); | if (client == null) throw new ArgumentNullException(nameof(client)); | ||||
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); | if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); | ||||
@@ -1,18 +1,16 @@ | |||||
using System; | using System; | ||||
using System.Collections.Generic; | using System.Collections.Generic; | ||||
using System.Threading.Tasks; | using System.Threading.Tasks; | ||||
using MQTTnet.Core.Client; | |||||
namespace MQTTnet.Core.Server | namespace MQTTnet.Core.Server | ||||
{ | { | ||||
public interface IMqttServer | |||||
public interface IMqttServer : IApplicationMessageReceiver, IApplicationMessagePublisher | |||||
{ | { | ||||
event EventHandler<MqttClientConnectedEventArgs> ClientConnected; | event EventHandler<MqttClientConnectedEventArgs> ClientConnected; | ||||
event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | ||||
event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; | |||||
IList<ConnectedMqttClient> GetConnectedClients(); | IList<ConnectedMqttClient> GetConnectedClients(); | ||||
void Publish(MqttApplicationMessage applicationMessage); | |||||
Task StartAsync(); | Task StartAsync(); | ||||
Task StopAsync(); | Task StopAsync(); | ||||
@@ -1,17 +0,0 @@ | |||||
using System; | |||||
namespace MQTTnet.Core.Server | |||||
{ | |||||
public sealed class MqttApplicationMessageReceivedEventArgs : EventArgs | |||||
{ | |||||
public MqttApplicationMessageReceivedEventArgs(string clientId, MqttApplicationMessage applicationMessage) | |||||
{ | |||||
ClientId = clientId; | |||||
ApplicationMessage = applicationMessage ?? throw new ArgumentNullException(nameof(applicationMessage)); | |||||
} | |||||
public string ClientId { get; } | |||||
public MqttApplicationMessage ApplicationMessage { get; } | |||||
} | |||||
} |
@@ -11,6 +11,7 @@ using MQTTnet.Core.Protocol; | |||||
using MQTTnet.Core.Serializer; | using MQTTnet.Core.Serializer; | ||||
using Microsoft.Extensions.Logging; | using Microsoft.Extensions.Logging; | ||||
using Microsoft.Extensions.Options; | using Microsoft.Extensions.Options; | ||||
using MQTTnet.Core.Client; | |||||
namespace MQTTnet.Core.Server | namespace MQTTnet.Core.Server | ||||
{ | { | ||||
@@ -28,9 +29,9 @@ namespace MQTTnet.Core.Server | |||||
_mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory)); | _mqttClientSesssionFactory = mqttClientSesssionFactory ?? throw new ArgumentNullException(nameof(mqttClientSesssionFactory)); | ||||
} | } | ||||
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; | |||||
public event EventHandler<MqttClientConnectedEventArgs> ClientConnected; | public event EventHandler<MqttClientConnectedEventArgs> ClientConnected; | ||||
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | ||||
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; | |||||
public MqttClientRetainedMessagesManager RetainedMessagesManager { get; } | public MqttClientRetainedMessagesManager RetainedMessagesManager { get; } | ||||
public MqttServerOptions Options { get; } | public MqttServerOptions Options { get; } | ||||
@@ -6,6 +6,7 @@ using MQTTnet.Core.Adapter; | |||||
using Microsoft.Extensions.Logging; | using Microsoft.Extensions.Logging; | ||||
using Microsoft.Extensions.Options; | using Microsoft.Extensions.Options; | ||||
using System.Linq; | using System.Linq; | ||||
using MQTTnet.Core.Client; | |||||
namespace MQTTnet.Core.Server | namespace MQTTnet.Core.Server | ||||
{ | { | ||||
@@ -45,12 +46,22 @@ namespace MQTTnet.Core.Server | |||||
public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | public event EventHandler<MqttClientDisconnectedEventArgs> ClientDisconnected; | ||||
public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; | public event EventHandler<MqttApplicationMessageReceivedEventArgs> ApplicationMessageReceived; | ||||
public void Publish(MqttApplicationMessage applicationMessage) | |||||
public Task PublishAsync(IEnumerable<MqttApplicationMessage> applicationMessages) | |||||
{ | { | ||||
if (applicationMessage == null) throw new ArgumentNullException(nameof(applicationMessage)); | |||||
if (applicationMessages == null) throw new ArgumentNullException(nameof(applicationMessages)); | |||||
_options.ApplicationMessageInterceptor?.Invoke(applicationMessage); | |||||
_clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); | |||||
if (_cancellationTokenSource == null) | |||||
{ | |||||
throw new InvalidOperationException("The server is not started."); | |||||
} | |||||
foreach (var applicationMessage in applicationMessages) | |||||
{ | |||||
_options.ApplicationMessageInterceptor?.Invoke(applicationMessage); | |||||
_clientSessionsManager.DispatchApplicationMessage(null, applicationMessage); | |||||
} | |||||
return Task.FromResult(0); | |||||
} | } | ||||
public async Task StartAsync() | public async Task StartAsync() | ||||
@@ -4,7 +4,7 @@ using MQTTnet.Core.Protocol; | |||||
namespace MQTTnet.Core.Server | namespace MQTTnet.Core.Server | ||||
{ | { | ||||
public sealed class MqttServerOptions | |||||
public class MqttServerOptions | |||||
{ | { | ||||
public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions(); | public MqttServerDefaultEndpointOptions DefaultEndpointOptions { get; } = new MqttServerDefaultEndpointOptions(); | ||||
@@ -122,7 +122,7 @@ namespace MQTTnet.Core.Tests | |||||
var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); | var message = new MqttApplicationMessageBuilder().WithTopic("a").WithAtLeastOnceQoS().Build(); | ||||
await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); | await c1.SubscribeAsync(new TopicFilter("a", MqttQualityOfServiceLevel.AtLeastOnce)); | ||||
s.Publish(message); | |||||
s.PublishAsync(message).Wait(); | |||||
await Task.Delay(500); | await Task.Delay(500); | ||||
await s.StopAsync(); | await s.StopAsync(); | ||||
@@ -180,7 +180,13 @@ namespace MQTTnet.Core.Tests | |||||
public async Task MqttServer_ClearRetainedMessage() | public async Task MqttServer_ClearRetainedMessage() | ||||
{ | { | ||||
var serverAdapter = new TestMqttServerAdapter(); | var serverAdapter = new TestMqttServerAdapter(); | ||||
var s = new MqttFactory().CreateMqttServer(); | |||||
var services = new ServiceCollection() | |||||
.AddLogging() | |||||
.AddMqttServer() // TODO: Is there maybe an easier way for the library user to set the options? | |||||
.AddSingleton<IMqttServerAdapter>(serverAdapter) | |||||
.BuildServiceProvider(); | |||||
var s = new MqttFactory(services).CreateMqttServer(); | |||||
await s.StartAsync(); | await s.StartAsync(); | ||||
var c1 = await serverAdapter.ConnectTestClient(s, "c1"); | var c1 = await serverAdapter.ConnectTestClient(s, "c1"); | ||||
@@ -207,11 +213,12 @@ namespace MQTTnet.Core.Tests | |||||
var serverAdapter = new TestMqttServerAdapter(); | var serverAdapter = new TestMqttServerAdapter(); | ||||
var services = new ServiceCollection() | var services = new ServiceCollection() | ||||
.AddMqttServer(options => options.Storage = storage) | |||||
.AddLogging() | |||||
.AddMqttServer(options => options.Storage = storage) // TODO: Is there maybe an easier way for the library user to set the options? | |||||
.AddSingleton<IMqttServerAdapter>(serverAdapter) | .AddSingleton<IMqttServerAdapter>(serverAdapter) | ||||
.BuildServiceProvider(); | .BuildServiceProvider(); | ||||
var s = new MqttFactory().CreateMqttServer(); | |||||
var s = new MqttFactory(services).CreateMqttServer(); // TODO: Like here? | |||||
await s.StartAsync(); | await s.StartAsync(); | ||||
var c1 = await serverAdapter.ConnectTestClient(s, "c1"); | var c1 = await serverAdapter.ConnectTestClient(s, "c1"); | ||||
@@ -246,6 +253,7 @@ namespace MQTTnet.Core.Tests | |||||
var serverAdapter = new TestMqttServerAdapter(); | var serverAdapter = new TestMqttServerAdapter(); | ||||
var services = new ServiceCollection() | var services = new ServiceCollection() | ||||
.AddLogging() | |||||
.AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor) | .AddMqttServer(options => options.ApplicationMessageInterceptor = Interceptor) | ||||
.AddSingleton<IMqttServerAdapter>(serverAdapter) | .AddSingleton<IMqttServerAdapter>(serverAdapter) | ||||
.BuildServiceProvider(); | .BuildServiceProvider(); | ||||
@@ -8,6 +8,7 @@ using Microsoft.Extensions.FileProviders; | |||||
using Microsoft.Extensions.Logging; | using Microsoft.Extensions.Logging; | ||||
using MQTTnet.AspnetCore; | using MQTTnet.AspnetCore; | ||||
using MQTTnet.Core; | using MQTTnet.Core; | ||||
using MQTTnet.Core.Client; | |||||
namespace MQTTnet.TestApp.AspNetCore2 | namespace MQTTnet.TestApp.AspNetCore2 | ||||
{ | { | ||||
@@ -29,7 +30,7 @@ namespace MQTTnet.TestApp.AspNetCore2 | |||||
while (true) | while (true) | ||||
{ | { | ||||
server.Publish(msg.Build()); | |||||
server.PublishAsync(msg.Build()).Wait(); | |||||
await Task.Delay(TimeSpan.FromSeconds(2)); | await Task.Delay(TimeSpan.FromSeconds(2)); | ||||
msg.WithPayload("Mqtt is still awesome at " + DateTime.Now); | msg.WithPayload("Mqtt is still awesome at " + DateTime.Now); | ||||
} | } | ||||