mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-18 10:26:11 +08:00
added aspnetcore package and adapted sample app
This commit is contained in:
@@ -0,0 +1,42 @@
|
||||
using System;
|
||||
using Microsoft.AspNetCore.Builder;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using MQTTnet.AspNetCore;
|
||||
using MQTTnet.Core.Server;
|
||||
|
||||
namespace MQTTnet.AspnetCore
|
||||
{
|
||||
public static class ApplicationBuilderExtensions
|
||||
{
|
||||
public static IApplicationBuilder UseMqttEndpoint(this IApplicationBuilder app, string path = "/mqtt")
|
||||
{
|
||||
app.UseWebSockets();
|
||||
app.Use(async (context, next) =>
|
||||
{
|
||||
if (context.Request.Path == path && context.WebSockets.IsWebSocketRequest)
|
||||
{
|
||||
var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
|
||||
using (var webSocket = await context.WebSockets.AcceptWebSocketAsync("mqtt"))
|
||||
{
|
||||
await adapter.AcceptWebSocketAsync(webSocket);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
await next();
|
||||
}
|
||||
});
|
||||
|
||||
return app;
|
||||
}
|
||||
|
||||
public static IApplicationBuilder UseMqttServer(this IApplicationBuilder app, Action<IMqttServer> configure)
|
||||
{
|
||||
var server = app.ApplicationServices.GetRequiredService<IMqttServer>();
|
||||
|
||||
configure(server);
|
||||
|
||||
return app;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>netstandard2.0</TargetFramework>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.AspNetCore.Http.Abstractions" Version="2.0.0" />
|
||||
<PackageReference Include="Microsoft.AspNetCore.WebSockets" Version="2.0.0" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.0.0" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\MQTTnet.Netstandard\MQTTnet.NetStandard.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,29 @@
|
||||
using System.Collections.Generic;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Server;
|
||||
|
||||
namespace MQTTnet.AspnetCore
|
||||
{
|
||||
public class MqttHostedServer : MqttServer, IHostedService
|
||||
{
|
||||
public MqttHostedServer(IOptions<MqttServerOptions> options, IEnumerable<IMqttServerAdapter> adapters, ILogger<MqttServer> logger, MqttClientSessionsManager clientSessionsManager)
|
||||
: base(options, adapters, logger, clientSessionsManager)
|
||||
{
|
||||
}
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return StartAsync();
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
return StopAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
using System;
|
||||
using System.IO;
|
||||
using System.Net.WebSockets;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Channel;
|
||||
using MQTTnet.Core.Server;
|
||||
using MQTTnet.Implementations;
|
||||
|
||||
namespace MQTTnet.AspNetCore
|
||||
{
|
||||
public class MqttWebSocketServerAdapter : IMqttServerAdapter, IDisposable
|
||||
{
|
||||
private readonly IMqttCommunicationAdapterFactory _mqttCommunicationAdapterFactory;
|
||||
|
||||
public MqttWebSocketServerAdapter(IMqttCommunicationAdapterFactory mqttCommunicationAdapterFactory)
|
||||
{
|
||||
_mqttCommunicationAdapterFactory = mqttCommunicationAdapterFactory ?? throw new ArgumentNullException(nameof(mqttCommunicationAdapterFactory));
|
||||
}
|
||||
|
||||
public event EventHandler<MqttServerAdapterClientAcceptedEventArgs> ClientAccepted;
|
||||
|
||||
public Task StartAsync(MqttServerOptions options)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync()
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task AcceptWebSocketAsync(WebSocket webSocket)
|
||||
{
|
||||
if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));
|
||||
|
||||
var channel = new MqttWebSocketServerChannel(webSocket);
|
||||
var clientAdapter = _mqttCommunicationAdapterFactory.CreateServerMqttCommunicationAdapter(channel);
|
||||
|
||||
var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter);
|
||||
ClientAccepted?.Invoke(this, eventArgs);
|
||||
return eventArgs.SessionTask;
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
StopAsync();
|
||||
}
|
||||
|
||||
private class MqttWebSocketServerChannel : IMqttCommunicationChannel, IDisposable
|
||||
{
|
||||
private readonly WebSocket _webSocket;
|
||||
|
||||
public MqttWebSocketServerChannel(WebSocket webSocket)
|
||||
{
|
||||
_webSocket = webSocket ?? throw new ArgumentNullException(nameof(webSocket));
|
||||
|
||||
RawReceiveStream = new WebSocketStream(_webSocket);
|
||||
}
|
||||
|
||||
public Stream SendStream => RawReceiveStream;
|
||||
public Stream ReceiveStream => RawReceiveStream;
|
||||
public Stream RawReceiveStream { get; }
|
||||
|
||||
public Task ConnectAsync()
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task DisconnectAsync()
|
||||
{
|
||||
RawReceiveStream?.Dispose();
|
||||
|
||||
if (_webSocket == null)
|
||||
{
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
return _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None);
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
RawReceiveStream?.Dispose();
|
||||
SendStream?.Dispose();
|
||||
ReceiveStream?.Dispose();
|
||||
|
||||
_webSocket?.Dispose();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using MQTTnet.AspNetCore;
|
||||
using MQTTnet.Core.Adapter;
|
||||
using MQTTnet.Core.Server;
|
||||
|
||||
namespace MQTTnet.AspnetCore
|
||||
{
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddHostedMqttServer(this IServiceCollection services)
|
||||
{
|
||||
services.AddMqttServerServices();
|
||||
|
||||
services.AddSingleton<IHostedService>(s => s.GetService<MqttHostedServer>());
|
||||
services.AddSingleton<IMqttServer>(s => s.GetService<MqttHostedServer>());
|
||||
services.AddSingleton<MqttHostedServer>();
|
||||
|
||||
services.AddSingleton<MqttWebSocketServerAdapter>();
|
||||
services.AddSingleton<IMqttServerAdapter>(s => s.GetService<MqttWebSocketServerAdapter>());
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -14,15 +14,21 @@ namespace MQTTnet
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
public static IServiceCollection AddMqttServer(this IServiceCollection services)
|
||||
{
|
||||
services.AddMqttServerServices();
|
||||
|
||||
services.AddSingleton<IMqttServer>(s => s.GetService<MqttServer>());
|
||||
services.AddSingleton<MqttServer>();
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
public static IServiceCollection AddMqttServerServices(this IServiceCollection services)
|
||||
{
|
||||
services.AddOptions();
|
||||
services.AddSingleton<MqttFactory>();
|
||||
services.AddSingleton<IMqttCommunicationAdapterFactory, MqttFactory>();
|
||||
services.AddSingleton<IMqttClientSesssionFactory, MqttFactory>();
|
||||
|
||||
|
||||
services.AddSingleton<IMqttServer,MqttServer>();
|
||||
services.AddSingleton<MqttServer>();
|
||||
services.AddSingleton<IMqttCommunicationAdapterFactory>(s => s.GetService<MqttFactory>());
|
||||
services.AddSingleton<IMqttClientSesssionFactory>(s => s.GetService<MqttFactory>());
|
||||
|
||||
services.AddTransient<IMqttServerAdapter, MqttServerAdapter>();
|
||||
services.AddTransient<IMqttPacketSerializer, MqttPacketSerializer>();
|
||||
@@ -42,7 +48,7 @@ namespace MQTTnet
|
||||
public static IServiceCollection AddMqttClient(this IServiceCollection services)
|
||||
{
|
||||
services.AddSingleton<MqttFactory>();
|
||||
services.AddSingleton<IMqttCommunicationAdapterFactory, MqttFactory>();
|
||||
services.AddSingleton<IMqttCommunicationAdapterFactory>(s => s.GetService<MqttFactory>());
|
||||
|
||||
services.AddTransient<IMqttClient, MqttClient>();
|
||||
services.AddTransient<MqttClient>();
|
||||
|
||||
Reference in New Issue
Block a user