mirror of
https://github.com/dotnet/MQTTnet.git
synced 2026-10-27 00:40:01 +08:00
Add new factory overloads.
This commit is contained in:
@@ -19,7 +19,11 @@
|
|||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1'">
|
<ItemGroup Condition="'$(TargetFramework)' != 'netcoreapp3.1'">
|
||||||
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.0.0" />
|
<PackageReference Include="Microsoft.AspNetCore.Http.Connections" Version="1.1.0" />
|
||||||
|
</ItemGroup>
|
||||||
|
|
||||||
|
<ItemGroup>
|
||||||
|
<PackageReference Include="System.IO.Pipelines" Version="4.7.2" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
|
|||||||
@@ -11,11 +11,18 @@ using System.Security.Cryptography.X509Certificates;
|
|||||||
using System.Threading;
|
using System.Threading;
|
||||||
using System.Threading.Tasks;
|
using System.Threading.Tasks;
|
||||||
using MQTTnet.AspNetCore.Extensions;
|
using MQTTnet.AspNetCore.Extensions;
|
||||||
|
using MQTTnet.Internal;
|
||||||
|
|
||||||
namespace MQTTnet.AspNetCore
|
namespace MQTTnet.AspNetCore
|
||||||
{
|
{
|
||||||
public class MqttConnectionContext : IMqttChannelAdapter
|
public sealed class MqttConnectionContext : IMqttChannelAdapter
|
||||||
{
|
{
|
||||||
|
readonly AsyncLock _writerLock = new AsyncLock();
|
||||||
|
readonly SpanBasedMqttPacketBodyReader _reader;
|
||||||
|
|
||||||
|
PipeReader _input;
|
||||||
|
PipeWriter _output;
|
||||||
|
|
||||||
public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection)
|
public MqttConnectionContext(MqttPacketFormatterAdapter packetFormatterAdapter, ConnectionContext connection)
|
||||||
{
|
{
|
||||||
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
|
PacketFormatterAdapter = packetFormatterAdapter ?? throw new ArgumentNullException(nameof(packetFormatterAdapter));
|
||||||
@@ -30,10 +37,6 @@ namespace MQTTnet.AspNetCore
|
|||||||
_reader = new SpanBasedMqttPacketBodyReader();
|
_reader = new SpanBasedMqttPacketBodyReader();
|
||||||
}
|
}
|
||||||
|
|
||||||
private PipeReader _input;
|
|
||||||
private PipeWriter _output;
|
|
||||||
private readonly SpanBasedMqttPacketBodyReader _reader;
|
|
||||||
|
|
||||||
public string Endpoint
|
public string Endpoint
|
||||||
{
|
{
|
||||||
get
|
get
|
||||||
@@ -58,9 +61,8 @@ namespace MQTTnet.AspNetCore
|
|||||||
|
|
||||||
public X509Certificate2 ClientCertificate => Http?.HttpContext?.Connection?.ClientCertificate;
|
public X509Certificate2 ClientCertificate => Http?.HttpContext?.Connection?.ClientCertificate;
|
||||||
|
|
||||||
private IHttpContextFeature Http => Connection.Features.Get<IHttpContextFeature>();
|
|
||||||
|
|
||||||
public ConnectionContext Connection { get; }
|
public ConnectionContext Connection { get; }
|
||||||
|
|
||||||
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
|
public MqttPacketFormatterAdapter PacketFormatterAdapter { get; }
|
||||||
|
|
||||||
public long BytesSent { get; set; }
|
public long BytesSent { get; set; }
|
||||||
@@ -69,7 +71,7 @@ namespace MQTTnet.AspNetCore
|
|||||||
public Action ReadingPacketStartedCallback { get; set; }
|
public Action ReadingPacketStartedCallback { get; set; }
|
||||||
public Action ReadingPacketCompletedCallback { get; set; }
|
public Action ReadingPacketCompletedCallback { get; set; }
|
||||||
|
|
||||||
private readonly SemaphoreSlim _writerSemaphore = new SemaphoreSlim(1, 1);
|
IHttpContextFeature Http => Connection.Features.Get<IHttpContextFeature>();
|
||||||
|
|
||||||
public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
public async Task ConnectAsync(TimeSpan timeout, CancellationToken cancellationToken)
|
||||||
{
|
{
|
||||||
@@ -165,8 +167,7 @@ namespace MQTTnet.AspNetCore
|
|||||||
var msg = buffer.AsMemory();
|
var msg = buffer.AsMemory();
|
||||||
var output = _output;
|
var output = _output;
|
||||||
|
|
||||||
await _writerSemaphore.WaitAsync(cancellationToken).ConfigureAwait(false);
|
using (await _writerLock.WaitAsync(cancellationToken).ConfigureAwait(false))
|
||||||
try
|
|
||||||
{
|
{
|
||||||
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
|
var result = await output.WriteAsync(msg, cancellationToken).ConfigureAwait(false);
|
||||||
if (result.IsCompleted)
|
if (result.IsCompleted)
|
||||||
@@ -176,10 +177,6 @@ namespace MQTTnet.AspNetCore
|
|||||||
|
|
||||||
PacketFormatterAdapter.FreeBuffer();
|
PacketFormatterAdapter.FreeBuffer();
|
||||||
}
|
}
|
||||||
finally
|
|
||||||
{
|
|
||||||
_writerSemaphore.Release();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public void Dispose()
|
public void Dispose()
|
||||||
|
|||||||
@@ -1,12 +1,29 @@
|
|||||||
using System.Collections.Generic;
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
using System.Linq;
|
using System.Linq;
|
||||||
|
using Microsoft.AspNetCore.Http;
|
||||||
|
|
||||||
namespace MQTTnet.AspNetCore
|
namespace MQTTnet.AspNetCore
|
||||||
{
|
{
|
||||||
public static class MqttSubProtocolSelector
|
public static class MqttSubProtocolSelector
|
||||||
{
|
{
|
||||||
|
public static string SelectSubProtocol(HttpRequest request)
|
||||||
|
{
|
||||||
|
if (request == null) throw new ArgumentNullException(nameof(request));
|
||||||
|
|
||||||
|
string subProtocol = null;
|
||||||
|
if (request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
|
||||||
|
{
|
||||||
|
subProtocol = SelectSubProtocol(requestedSubProtocolValues);
|
||||||
|
}
|
||||||
|
|
||||||
|
return subProtocol;
|
||||||
|
}
|
||||||
|
|
||||||
public static string SelectSubProtocol(IList<string> requestedSubProtocolValues)
|
public static string SelectSubProtocol(IList<string> requestedSubProtocolValues)
|
||||||
{
|
{
|
||||||
|
if (requestedSubProtocolValues == null) throw new ArgumentNullException(nameof(requestedSubProtocolValues));
|
||||||
|
|
||||||
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
|
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
|
||||||
return requestedSubProtocolValues
|
return requestedSubProtocolValues
|
||||||
.OrderByDescending(p => p.Length)
|
.OrderByDescending(p => p.Length)
|
||||||
|
|||||||
@@ -39,12 +39,12 @@
|
|||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<PackageReference Include="IronPython" Version="2.7.10-candidate1" />
|
<PackageReference Include="IronPython" Version="2.7.10" />
|
||||||
<PackageReference Include="IronPython.StdLib" Version="2.7.10-candidate1" />
|
<PackageReference Include="IronPython.StdLib" Version="2.7.10" />
|
||||||
<PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="3.1.3" />
|
<PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="3.1.4" />
|
||||||
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
||||||
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.3.3" />
|
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerGen" Version="5.4.1" />
|
||||||
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.3.3" />
|
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="5.4.1" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
using MQTTnet.Adapter;
|
using System;
|
||||||
|
using MQTTnet.Adapter;
|
||||||
using MQTTnet.Diagnostics;
|
using MQTTnet.Diagnostics;
|
||||||
|
using MQTTnet.LowLevelClient;
|
||||||
|
|
||||||
namespace MQTTnet.Client
|
namespace MQTTnet.Client
|
||||||
{
|
{
|
||||||
@@ -7,6 +9,14 @@ namespace MQTTnet.Client
|
|||||||
{
|
{
|
||||||
IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory);
|
IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory);
|
||||||
|
|
||||||
|
ILowLevelMqttClient CreateLowLevelMqttClient();
|
||||||
|
|
||||||
|
ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger);
|
||||||
|
|
||||||
|
ILowLevelMqttClient CreateLowLevelMqttClient(IMqttClientAdapterFactory clientAdapterFactory);
|
||||||
|
|
||||||
|
ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory);
|
||||||
|
|
||||||
IMqttClient CreateMqttClient();
|
IMqttClient CreateMqttClient();
|
||||||
|
|
||||||
IMqttClient CreateMqttClient(IMqttNetLogger logger);
|
IMqttClient CreateMqttClient(IMqttNetLogger logger);
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
using MQTTnet.Client;
|
using System.Collections.Generic;
|
||||||
|
using MQTTnet.Client;
|
||||||
using MQTTnet.Diagnostics;
|
using MQTTnet.Diagnostics;
|
||||||
using MQTTnet.Server;
|
using MQTTnet.Server;
|
||||||
|
|
||||||
@@ -7,5 +8,7 @@ namespace MQTTnet
|
|||||||
public interface IMqttFactory : IMqttClientFactory, IMqttServerFactory
|
public interface IMqttFactory : IMqttClientFactory, IMqttServerFactory
|
||||||
{
|
{
|
||||||
IMqttNetLogger DefaultLogger { get; }
|
IMqttNetLogger DefaultLogger { get; }
|
||||||
|
|
||||||
|
IDictionary<object, object> Properties { get; }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,10 +5,10 @@ using System.Threading.Tasks;
|
|||||||
namespace MQTTnet.Internal
|
namespace MQTTnet.Internal
|
||||||
{
|
{
|
||||||
// From Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/12/building-async-coordination-primitives-part-6-asynclock/)
|
// From Stephen Toub (https://blogs.msdn.microsoft.com/pfxteam/2012/02/12/building-async-coordination-primitives-part-6-asynclock/)
|
||||||
public class AsyncLock : IDisposable
|
public sealed class AsyncLock : IDisposable
|
||||||
{
|
{
|
||||||
private readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
|
readonly SemaphoreSlim _semaphore = new SemaphoreSlim(1, 1);
|
||||||
private readonly Task<IDisposable> _releaser;
|
readonly Task<IDisposable> _releaser;
|
||||||
|
|
||||||
public AsyncLock()
|
public AsyncLock()
|
||||||
{
|
{
|
||||||
@@ -39,9 +39,9 @@ namespace MQTTnet.Internal
|
|||||||
_semaphore?.Dispose();
|
_semaphore?.Dispose();
|
||||||
}
|
}
|
||||||
|
|
||||||
private class Releaser : IDisposable
|
class Releaser : IDisposable
|
||||||
{
|
{
|
||||||
private readonly AsyncLock _toRelease;
|
readonly AsyncLock _toRelease;
|
||||||
|
|
||||||
internal Releaser(AsyncLock toRelease)
|
internal Releaser(AsyncLock toRelease)
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ using MQTTnet.LowLevelClient;
|
|||||||
using MQTTnet.Server;
|
using MQTTnet.Server;
|
||||||
using System;
|
using System;
|
||||||
using System.Collections.Generic;
|
using System.Collections.Generic;
|
||||||
|
using System.Linq;
|
||||||
|
|
||||||
namespace MQTTnet
|
namespace MQTTnet
|
||||||
{
|
{
|
||||||
@@ -25,6 +26,13 @@ namespace MQTTnet
|
|||||||
|
|
||||||
public IMqttNetLogger DefaultLogger { get; }
|
public IMqttNetLogger DefaultLogger { get; }
|
||||||
|
|
||||||
|
public IList<Func<IMqttFactory, IMqttServerAdapter>> DefaultServerAdapters { get; } = new List<Func<IMqttFactory, IMqttServerAdapter>>
|
||||||
|
{
|
||||||
|
factory => new MqttTcpServerAdapter(factory.DefaultLogger)
|
||||||
|
};
|
||||||
|
|
||||||
|
public IDictionary<object, object> Properties { get; } = new Dictionary<object, object>();
|
||||||
|
|
||||||
public IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory)
|
public IMqttFactory UseClientAdapterFactory(IMqttClientAdapterFactory clientAdapterFactory)
|
||||||
{
|
{
|
||||||
_clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory));
|
_clientAdapterFactory = clientAdapterFactory ?? throw new ArgumentNullException(nameof(clientAdapterFactory));
|
||||||
@@ -50,10 +58,10 @@ namespace MQTTnet
|
|||||||
return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger);
|
return new LowLevelMqttClient(_clientAdapterFactory, DefaultLogger);
|
||||||
}
|
}
|
||||||
|
|
||||||
public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactoryy)
|
public ILowLevelMqttClient CreateLowLevelMqttClient(IMqttNetLogger logger, IMqttClientAdapterFactory clientAdapterFactory)
|
||||||
{
|
{
|
||||||
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
||||||
if (clientAdapterFactoryy == null) throw new ArgumentNullException(nameof(clientAdapterFactoryy));
|
if (clientAdapterFactory == null) throw new ArgumentNullException(nameof(clientAdapterFactory));
|
||||||
|
|
||||||
return new LowLevelMqttClient(_clientAdapterFactory, logger);
|
return new LowLevelMqttClient(_clientAdapterFactory, logger);
|
||||||
}
|
}
|
||||||
@@ -94,7 +102,8 @@ namespace MQTTnet
|
|||||||
{
|
{
|
||||||
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
||||||
|
|
||||||
return CreateMqttServer(new List<IMqttServerAdapter> { new MqttTcpServerAdapter(logger) }, logger);
|
var serverAdapters = DefaultServerAdapters.Select(a => a.Invoke(this));
|
||||||
|
return CreateMqttServer(serverAdapters, logger);
|
||||||
}
|
}
|
||||||
|
|
||||||
public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters, IMqttNetLogger logger)
|
public IMqttServer CreateMqttServer(IEnumerable<IMqttServerAdapter> serverAdapters, IMqttNetLogger logger)
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
using System.Collections.Generic;
|
using System;
|
||||||
|
using System.Collections.Generic;
|
||||||
using MQTTnet.Adapter;
|
using MQTTnet.Adapter;
|
||||||
using MQTTnet.Diagnostics;
|
using MQTTnet.Diagnostics;
|
||||||
|
|
||||||
@@ -6,6 +7,8 @@ namespace MQTTnet.Server
|
|||||||
{
|
{
|
||||||
public interface IMqttServerFactory
|
public interface IMqttServerFactory
|
||||||
{
|
{
|
||||||
|
IList<Func<IMqttFactory, IMqttServerAdapter>> DefaultServerAdapters { get; }
|
||||||
|
|
||||||
IMqttServer CreateMqttServer();
|
IMqttServer CreateMqttServer();
|
||||||
|
|
||||||
IMqttServer CreateMqttServer(IMqttNetLogger logger);
|
IMqttServer CreateMqttServer(IMqttNetLogger logger);
|
||||||
|
|||||||
@@ -6,7 +6,7 @@
|
|||||||
</PropertyGroup>
|
</PropertyGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.0" />
|
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.1" />
|
||||||
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
||||||
<PackageReference Include="MSTest.TestFramework" Version="2.1.1" />
|
<PackageReference Include="MSTest.TestFramework" Version="2.1.1" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|||||||
@@ -10,7 +10,7 @@
|
|||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<PackageReference Include="BenchmarkDotNet" Version="0.12.1" />
|
<PackageReference Include="BenchmarkDotNet" Version="0.12.1" />
|
||||||
<PackageReference Include="System.IO.Pipelines" Version="4.7.1" />
|
<PackageReference Include="System.IO.Pipelines" Version="4.7.2" />
|
||||||
<PackageReference Include="Microsoft.AspNetCore" Version="2.2.0" />
|
<PackageReference Include="Microsoft.AspNetCore" Version="2.2.0" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
|
|||||||
@@ -8,7 +8,7 @@
|
|||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
<PackageReference Include="MSTest.TestAdapter" Version="2.1.1" />
|
||||||
<PackageReference Include="MSTest.TestFramework" Version="2.1.1" />
|
<PackageReference Include="MSTest.TestFramework" Version="2.1.1" />
|
||||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.0" />
|
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.6.1" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
<ItemGroup>
|
||||||
|
|||||||
Reference in New Issue
Block a user