Nov 7, 2024 · Nov 12, 2024 · Nov 13, 2024 · Nov 13, 2024 · Nov 13, 2024 · Nov 13, 2024
diff --git a/Samples/MQTTnet.Samples.csproj b/Samples/MQTTnet.Samples.csproj
        <NuGetAuditMode>all</NuGetAuditMode>
        <NuGetAudit>true</NuGetAudit>
        <NuGetAuditLevel>low</NuGetAuditLevel>
        <AnalysisLevel>latest-Recommended</AnalysisLevel>
        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
    </PropertyGroup>

    <ItemGroup>
diff --git a/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj b/Source/MQTTnet.AspTestApp/MQTTnet.AspTestApp.csproj
        <NuGetAuditMode>all</NuGetAuditMode>
        <NuGetAudit>true</NuGetAudit>
        <NuGetAuditLevel>low</NuGetAuditLevel>
        <AnalysisLevel>latest-Recommended</AnalysisLevel>
        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
    </PropertyGroup>

    <ItemGroup>
diff --git a/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj b/Source/MQTTnet.AspnetCore/MQTTnet.AspNetCore.csproj
        <NuGetAudit>true</NuGetAudit>
        <NuGetAuditLevel>low</NuGetAuditLevel>
        <NuGetAuditLevel>low</NuGetAuditLevel>
        <AnalysisLevel>latest-Recommended</AnalysisLevel>
        <!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
    </PropertyGroup>

    <ItemGroup>
diff --git a/Source/MQTTnet.AspnetCore/MqttClientConnectionContextFactory.cs b/Source/MQTTnet.AspnetCore/MqttClientConnectionContextFactory.cs
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.

 usingSystem;
 usingMicrosoft.AspNetCore.Connections;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Formatter;
 using System;
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
 using System.Threading.Tasks;

 namespace MQTTnet.AspNetCore
 {
    public sealed class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
    {
        public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
        private readonly IConnectionFactory connectionFactory;

        public MqttClientConnectionContextFactory(IConnectionFactory connectionFactory)
        {
            this.connectionFactory = connectionFactory;
        }

        public async ValueTask<IMqttChannelAdapter> CreateClientAdapterAsync(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
        {
            if (options == null) throw new ArgumentNullException(nameof(options));

            switch (options.ChannelOptions)
            {
                case MqttClientTcpOptions tcpOptions:
                    {
                        vartcpConnection =new SocketConnection(tcpOptions.RemoteEndpoint);

                        varendPoint =await CreateIPEndPointAsync(tcpOptions.RemoteEndpoint);
                        var tcpConnection = await connectionFactory.ConnectAsync(endPoint);
                        var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535));
                        return new MqttConnectionContext(formatter, tcpConnection);
                    }
                    }
            }
        }

        private static async ValueTask<IPEndPoint> CreateIPEndPointAsync(EndPoint endpoint)
        {
            if (endpoint is IPEndPoint ipEndPoint)
            {
                return ipEndPoint;
            }

            if (endpoint is DnsEndPoint dnsEndPoint)
            {
                var hostEntry = await Dns.GetHostEntryAsync(dnsEndPoint.Host);
                var address = hostEntry.AddressList.OrderBy(item => item.AddressFamily).FirstOrDefault();
                return address == null
                    ? throw new SocketException((int)SocketError.HostNotFound)
                    : new IPEndPoint(address, dnsEndPoint.Port);
            }

            throw new NotSupportedException("Only supports IPEndPoint or DnsEndPoint for now.");
        }
    }
 }
diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs b/Source/MQTTnet.AspnetCore/MqttConnectionContext.cs
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.

 using System;
 using System.Buffers;
 using System.IO.Pipelines;
 using System.Net;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.AspNetCore.Connections;
 using Microsoft.AspNetCore.Http.Connections.Features;
 using Microsoft.AspNetCore.Http.Features;
 using MQTTnet.Formatter;
 using MQTTnet.Internal;
 using MQTTnet.Packets;
 using System;
 using System.Buffers;
 using System.IO.Pipelines;
 using System.Net;
 using System.Security.Cryptography.X509Certificates;
 using System.Threading;
 using System.Threading.Tasks;

 namespace MQTTnet.AspNetCore;

    readonly ConnectionContext _connection;
    readonly AsyncLock _writerLock = new();

    PipeReader _input;
    PipeWriter _output;
 readonlyPipeReader _input;
 readonlyPipeWriter _output;

    public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection)
    {
        PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
        _connection = connection ?? throw new ArgumentNullException(nameof(connection));

        if (!(_connection is SocketConnection tcp) || tcp.IsConnected)
        {
            _input = connection.Transport.Input;
            _output = connection.Transport.Output;
        }
        _input = connection.Transport.Input;
        _output = connection.Transport.Output;
    }

    public long BytesReceived { get; private set; }

    public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }

    publicasyncTask ConnectAsync(CancellationToken cancellationToken)
    public Task ConnectAsync(CancellationToken cancellationToken)
    {
        if (_connection is SocketConnection tcp && !tcp.IsConnected)
        {
            await tcp.StartAsync().ConfigureAwait(false);
        }

        _input = _connection.Transport.Input;
        _output = _connection.Transport.Output;
        return Task.CompletedTask;
    }

    public Task DisconnectAsync(CancellationToken cancellationToken)
    }

    public void Dispose()
    {
    {
        _writerLock.Dispose();
    }

diff --git a/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs b/Source/MQTTnet.AspnetCore/MqttConnectionHandler.cs
        }

        var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax));
        using (var adapter = new MqttConnectionContext(formatter, connection))
        using var adapter = new MqttConnectionContext(formatter, connection);
        var clientHandler = ClientHandler;
        if (clientHandler != null)
        {
            var clientHandler = ClientHandler;
            if (clientHandler != null)
            {
                await clientHandler(adapter).ConfigureAwait(false);
            }
            await clientHandler(adapter).ConfigureAwait(false);
        }
    }

diff --git a/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs b/Source/MQTTnet.AspnetCore/ServiceCollectionExtensions.cs
 // The .NET Foundation licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for more information.

 usingSystem;
 usingMicrosoft.AspNetCore.Connections;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Hosting;
 using MQTTnet.Adapter;
 using MQTTnet.Diagnostics.Logger;
 using MQTTnet.Server;
 using MQTTnet.Server.Internal.Adapter;
 using System;
 using System.Diagnostics.CodeAnalysis;
 using System.Reflection;

 namespace MQTTnet.AspNetCore;

 public static class ServiceCollectionExtensions
 {
    const string SocketConnectionFactoryTypeName = "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactory";
    const string SocketConnectionFactoryAssemblyName = "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets";

    public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
    {
        ArgumentNullException.ThrowIfNull(services);

        return services;
    }


    [DynamicDependency(DynamicallyAccessedMemberTypes.All, SocketConnectionFactoryTypeName, SocketConnectionFactoryAssemblyName)]
    public static IServiceCollection AddMqttClientConnectionContextFactory(this IServiceCollection services)
    {
        var socketConnectionFactoryType = Assembly.Load(SocketConnectionFactoryAssemblyName).GetType(SocketConnectionFactoryTypeName);
        services.AddSingleton(typeof(IConnectionFactory), socketConnectionFactoryType);
        services.TryAddSingleton<MqttClientConnectionContextFactory>();
        services.TryAddSingleton<IMqttClientAdapterFactory>(serviceProvider => serviceProvider.GetRequiredService<MqttClientConnectionContextFactory>());
        return services;
    }
 }
diff --git a/Source/MQTTnet.AspnetCore/SocketAwaitable.cs b/Source/MQTTnet.AspnetCore/SocketAwaitable.cs
Original file line number	Diff line number	Diff line change
Expand Up		@@ -14,7 +14,7 @@
		<NuGetAuditMode>all</NuGetAuditMode>
		<NuGetAudit>true</NuGetAudit>
		<NuGetAuditLevel>low</NuGetAuditLevel>
		<AnalysisLevel>latest-Recommended</AnalysisLevel>
		<!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
		</PropertyGroup>

		<ItemGroup>
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -13,7 +13,7 @@
		<NuGetAuditMode>all</NuGetAuditMode>
		<NuGetAudit>true</NuGetAudit>
		<NuGetAuditLevel>low</NuGetAuditLevel>
		<AnalysisLevel>latest-Recommended</AnalysisLevel>
		<!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
		</PropertyGroup>

		<ItemGroup>
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -37,7 +37,7 @@
		<NuGetAudit>true</NuGetAudit>
		<NuGetAuditLevel>low</NuGetAuditLevel>
		<NuGetAuditLevel>low</NuGetAuditLevel>
		<AnalysisLevel>latest-Recommended</AnalysisLevel>
		<!--<AnalysisLevel>latest-Recommended</AnalysisLevel>-->
		</PropertyGroup>

		<ItemGroup>
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -2,25 +2,37 @@
		// The .NET Foundation licenses this file to you under the MIT license.
		// See the LICENSE file in the project root for more information.

		usingSystem;
		usingMicrosoft.AspNetCore.Connections;
		using MQTTnet.Adapter;
		using MQTTnet.Diagnostics.Logger;
		using MQTTnet.Formatter;
		using System;
		using System.Linq;
		using System.Net;
		using System.Net.Sockets;
		using System.Threading.Tasks;

		namespace MQTTnet.AspNetCore
		{
		public sealed class MqttClientConnectionContextFactory : IMqttClientAdapterFactory
		{
		public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
		private readonly IConnectionFactory connectionFactory;

		public MqttClientConnectionContextFactory(IConnectionFactory connectionFactory)
		{
		this.connectionFactory = connectionFactory;
		}

		public async ValueTask<IMqttChannelAdapter> CreateClientAdapterAsync(MqttClientOptions options, MqttPacketInspector packetInspector, IMqttNetLogger logger)
		{
		if (options == null) throw new ArgumentNullException(nameof(options));

		switch (options.ChannelOptions)
		{
		case MqttClientTcpOptions tcpOptions:
		{
		vartcpConnection =new SocketConnection(tcpOptions.RemoteEndpoint);

		varendPoint =await CreateIPEndPointAsync(tcpOptions.RemoteEndpoint);
		var tcpConnection = await connectionFactory.ConnectAsync(endPoint);
		var formatter = new MqttPacketFormatterAdapter(options.ProtocolVersion, new MqttBufferWriter(4096, 65535));
		return new MqttConnectionContext(formatter, tcpConnection);
		}
Expand All		@@ -30,5 +42,24 @@ public IMqttChannelAdapter CreateClientAdapter(MqttClientOptions options, MqttPa
		}
		}
		}

		private static async ValueTask<IPEndPoint> CreateIPEndPointAsync(EndPoint endpoint)
		{
		if (endpoint is IPEndPoint ipEndPoint)
		{
		return ipEndPoint;
		}

		if (endpoint is DnsEndPoint dnsEndPoint)
		{
		var hostEntry = await Dns.GetHostEntryAsync(dnsEndPoint.Host);
		var address = hostEntry.AddressList.OrderBy(item => item.AddressFamily).FirstOrDefault();
		return address == null
		? throw new SocketException((int)SocketError.HostNotFound)
		: new IPEndPoint(address, dnsEndPoint.Port);
		}

		throw new NotSupportedException("Only supports IPEndPoint or DnsEndPoint for now.");
		}
		}
		}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -2,13 +2,6 @@
		// The .NET Foundation licenses this file to you under the MIT license.
		// See the LICENSE file in the project root for more information.

		using System;
		using System.Buffers;
		using System.IO.Pipelines;
		using System.Net;
		using System.Security.Cryptography.X509Certificates;
		using System.Threading;
		using System.Threading.Tasks;
		using Microsoft.AspNetCore.Connections;
		using Microsoft.AspNetCore.Http.Connections.Features;
		using Microsoft.AspNetCore.Http.Features;
Expand All		@@ -17,6 +10,13 @@
		using MQTTnet.Formatter;
		using MQTTnet.Internal;
		using MQTTnet.Packets;
		using System;
		using System.Buffers;
		using System.IO.Pipelines;
		using System.Net;
		using System.Security.Cryptography.X509Certificates;
		using System.Threading;
		using System.Threading.Tasks;

		namespace MQTTnet.AspNetCore;

Expand All		@@ -25,19 +25,15 @@ public sealed class MqttConnectionContext : IMqttChannelAdapter
		readonly ConnectionContext _connection;
		readonly AsyncLock _writerLock = new();

		PipeReader _input;
		PipeWriter _output;
		readonlyPipeReader _input;
		readonlyPipeWriter _output;

		public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection)
		{
		PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
		_connection = connection ?? throw new ArgumentNullException(nameof(connection));

		if (!(_connection is SocketConnection tcp) \|\| tcp.IsConnected)
		{
		_input = connection.Transport.Input;
		_output = connection.Transport.Output;
		}
		_input = connection.Transport.Input;
		_output = connection.Transport.Output;
		}

		public long BytesReceived { get; private set; }
Expand DownExpand Up		@@ -106,15 +102,9 @@ public bool IsSecureConnection

		public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }

		publicasyncTask ConnectAsync(CancellationToken cancellationToken)
		public Task ConnectAsync(CancellationToken cancellationToken)
		{
		if (_connection is SocketConnection tcp && !tcp.IsConnected)
		{
		await tcp.StartAsync().ConfigureAwait(false);
		}

		_input = _connection.Transport.Input;
		_output = _connection.Transport.Output;
		return Task.CompletedTask;
		}

		public Task DisconnectAsync(CancellationToken cancellationToken)
Expand All		@@ -126,7 +116,7 @@ public Task DisconnectAsync(CancellationToken cancellationToken)
		}

		public void Dispose()
		{
		{
		_writerLock.Dispose();
		}

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -33,13 +33,11 @@ public override async Task OnConnectedAsync(ConnectionContext connection)
		}

		var formatter = new MqttPacketFormatterAdapter(new MqttBufferWriter(_serverOptions.WriterBufferSize, _serverOptions.WriterBufferSizeMax));
		using (var adapter = new MqttConnectionContext(formatter, connection))
		using var adapter = new MqttConnectionContext(formatter, connection);
		var clientHandler = ClientHandler;
		if (clientHandler != null)
		{
		var clientHandler = ClientHandler;
		if (clientHandler != null)
		{
		await clientHandler(adapter).ConfigureAwait(false);
		}
		await clientHandler(adapter).ConfigureAwait(false);
		}
		}

Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -2,18 +2,25 @@
		// The .NET Foundation licenses this file to you under the MIT license.
		// See the LICENSE file in the project root for more information.

		usingSystem;
		usingMicrosoft.AspNetCore.Connections;
		using Microsoft.Extensions.DependencyInjection;
		using Microsoft.Extensions.DependencyInjection.Extensions;
		using Microsoft.Extensions.Hosting;
		using MQTTnet.Adapter;
		using MQTTnet.Diagnostics.Logger;
		using MQTTnet.Server;
		using MQTTnet.Server.Internal.Adapter;
		using System;
		using System.Diagnostics.CodeAnalysis;
		using System.Reflection;

		namespace MQTTnet.AspNetCore;

		public static class ServiceCollectionExtensions
		{
		const string SocketConnectionFactoryTypeName = "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets.SocketConnectionFactory";
		const string SocketConnectionFactoryAssemblyName = "Microsoft.AspNetCore.Server.Kestrel.Transport.Sockets";

		public static IServiceCollection AddHostedMqttServer(this IServiceCollection services, MqttServerOptions options)
		{
		ArgumentNullException.ThrowIfNull(services);
Expand DownExpand Up		@@ -106,4 +113,15 @@ public static IServiceCollection AddMqttWebSocketServerAdapter(this IServiceColl

		return services;
		}


		[DynamicDependency(DynamicallyAccessedMemberTypes.All, SocketConnectionFactoryTypeName, SocketConnectionFactoryAssemblyName)]
		public static IServiceCollection AddMqttClientConnectionContextFactory(this IServiceCollection services)
		{
		var socketConnectionFactoryType = Assembly.Load(SocketConnectionFactoryAssemblyName).GetType(SocketConnectionFactoryTypeName);
		services.AddSingleton(typeof(IConnectionFactory), socketConnectionFactoryType);
		services.TryAddSingleton<MqttClientConnectionContextFactory>();
		services.TryAddSingleton<IMqttClientAdapterFactory>(serviceProvider => serviceProvider.GetRequiredService<MqttClientConnectionContextFactory>());
		return services;
		}
		}