Refactor ASP.NET integration.

This commit is contained in:
Christian
2018-05-13 20:59:00 +02:00
parent dc7b9414d9
commit 3e6edb92c5
5 changed files with 34 additions and 34 deletions
@@ -13,32 +13,26 @@ namespace MQTTnet.AspNetCore
app.UseWebSockets(); app.UseWebSockets();
app.Use(async (context, next) => app.Use(async (context, next) =>
{ {
if (context.Request.Path == path && context.WebSockets.IsWebSocketRequest) if (!context.WebSockets.IsWebSocketRequest || context.Request.Path != path)
{
string subprotocol = null;
if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
{
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
subprotocol = requestedSubProtocolValues
.OrderByDescending(p => p.Length)
.FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
}
if (subprotocol == null)
{
return;
}
var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subprotocol))
{
await adapter.AcceptWebSocketAsync(webSocket);
}
}
else
{ {
await next(); await next();
return;
}
string subProtocol = null;
if (context.Request.Headers.TryGetValue("Sec-WebSocket-Protocol", out var requestedSubProtocolValues))
{
// Order the protocols to also match "mqtt", "mqttv-3.1", "mqttv-3.11" etc.
subProtocol = requestedSubProtocolValues
.OrderByDescending(p => p.Length)
.FirstOrDefault(p => p.ToLower().StartsWith("mqtt"));
}
var adapter = app.ApplicationServices.GetRequiredService<MqttWebSocketServerAdapter>();
using (var webSocket = await context.WebSockets.AcceptWebSocketAsync(subProtocol))
{
await adapter.AcceptWebSocketAsync(webSocket);
} }
}); });
@@ -23,7 +23,7 @@ namespace MQTTnet.AspNetCore
return Task.CompletedTask; return Task.CompletedTask;
} }
public Task AcceptWebSocketAsync(WebSocket webSocket) public async Task AcceptWebSocketAsync(WebSocket webSocket)
{ {
if (webSocket == null) throw new ArgumentNullException(nameof(webSocket)); if (webSocket == null) throw new ArgumentNullException(nameof(webSocket));
@@ -31,7 +31,11 @@ namespace MQTTnet.AspNetCore
var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter); var eventArgs = new MqttServerAdapterClientAcceptedEventArgs(clientAdapter);
ClientAccepted?.Invoke(this, eventArgs); ClientAccepted?.Invoke(this, eventArgs);
return eventArgs.SessionTask ?? Task.CompletedTask;
if (eventArgs.SessionTask != null)
{
await eventArgs.SessionTask.ConfigureAwait(false);
}
} }
public void Dispose() public void Dispose()
@@ -81,14 +81,14 @@ namespace MQTTnet.Server
_cancellationTokenSource = new CancellationTokenSource(); _cancellationTokenSource = new CancellationTokenSource();
_retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger); _retainedMessagesManager = new MqttRetainedMessagesManager(Options, _logger);
await _retainedMessagesManager.LoadMessagesAsync(); await _retainedMessagesManager.LoadMessagesAsync().ConfigureAwait(false);
_clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _logger); _clientSessionsManager = new MqttClientSessionsManager(Options, this, _retainedMessagesManager, _logger);
foreach (var adapter in _adapters) foreach (var adapter in _adapters)
{ {
adapter.ClientAccepted += OnClientAccepted; adapter.ClientAccepted += OnClientAccepted;
await adapter.StartAsync(Options); await adapter.StartAsync(Options).ConfigureAwait(false);
} }
_logger.Info("Started."); _logger.Info("Started.");
@@ -110,10 +110,10 @@ namespace MQTTnet.Server
foreach (var adapter in _adapters) foreach (var adapter in _adapters)
{ {
adapter.ClientAccepted -= OnClientAccepted; adapter.ClientAccepted -= OnClientAccepted;
await adapter.StopAsync(); await adapter.StopAsync().ConfigureAwait(false);
} }
await _clientSessionsManager.StopAsync(); await _clientSessionsManager.StopAsync().ConfigureAwait(false);
_logger.Info("Stopped."); _logger.Info("Stopped.");
} }
@@ -2,7 +2,7 @@
var client = connect('ws://' + location.host + '/mqtt', var client = connect('ws://' + location.host + '/mqtt',
{ {
clientId: "client1", clientId: "client" + Math.floor(Math.random() * 6) + 1
}); });
window.onbeforeunload = () => { window.onbeforeunload = () => {
@@ -23,11 +23,12 @@ publishButton.onclick = click => {
client.on('connect', () => { client.on('connect', () => {
client.subscribe('#', { qos: 0 }, (err, granted) => { client.subscribe('#', { qos: 0 }, (err, granted) => {
console.log(err); console.log(err);
}); });
client.publish('presence', 'Hello mqtt'); client.publish('presence', 'Hello mqtt');
stateParagraph.innerText = "connected"; stateParagraph.innerText = "connected";
showMsg("[connect]");
}); });
client.on("error", e => { client.on("error", e => {
@@ -36,6 +37,7 @@ client.on("error", e => {
client.on("reconnect", () => { client.on("reconnect", () => {
stateParagraph.innerText = "reconnecting"; stateParagraph.innerText = "reconnecting";
showMsg("[reconnect]");
}); });
client.on('message', (topic, message) => { client.on('message', (topic, message) => {
@@ -50,7 +52,7 @@ function showMsg(msg: string) {
msgsList.appendChild(node); msgsList.appendChild(node);
if (msgsList.childElementCount > 20) { if (msgsList.childElementCount > 50) {
msgsList.removeChild(msgsList.childNodes[0]); msgsList.removeChild(msgsList.childNodes[0]);
} }
} }
@@ -138,7 +138,7 @@
</ItemGroup> </ItemGroup>
<ItemGroup> <ItemGroup>
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform"> <PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform">
<Version>6.0.8</Version> <Version>6.1.4</Version>
</PackageReference> </PackageReference>
<PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls"> <PackageReference Include="Microsoft.Toolkit.Uwp.UI.Controls">
<Version>2.2.0</Version> <Version>2.2.0</Version>