Centralizing serializing hub messages
This commit is contained in:
parent
a0e490e549
commit
ad4784dbd2
|
|
@ -1,7 +1,6 @@
|
||||||
// Copyright (c) .NET Foundation. All rights reserved.
|
// Copyright (c) .NET Foundation. All rights reserved.
|
||||||
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
|
// Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
|
||||||
|
|
||||||
using System;
|
|
||||||
using System.IO;
|
using System.IO;
|
||||||
|
|
||||||
namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
|
namespace Microsoft.AspNetCore.SignalR.Internal.Protocol
|
||||||
|
|
|
||||||
|
|
@ -314,11 +314,9 @@ namespace Microsoft.AspNetCore.SignalR.Redis
|
||||||
|
|
||||||
private async Task WriteAsync(HubConnectionContext connection, HubMessage hubMessage)
|
private async Task WriteAsync(HubConnectionContext connection, HubMessage hubMessage)
|
||||||
{
|
{
|
||||||
var data = connection.Protocol.WriteToArray(hubMessage);
|
|
||||||
|
|
||||||
while (await connection.Output.WaitToWriteAsync())
|
while (await connection.Output.WaitToWriteAsync())
|
||||||
{
|
{
|
||||||
if (connection.Output.TryWrite(data))
|
if (connection.Output.TryWrite(hubMessage))
|
||||||
{
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -125,11 +125,9 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
|
|
||||||
private async Task WriteAsync(HubConnectionContext connection, HubMessage hubMessage)
|
private async Task WriteAsync(HubConnectionContext connection, HubMessage hubMessage)
|
||||||
{
|
{
|
||||||
var payload = connection.Protocol.WriteToArray(hubMessage);
|
|
||||||
|
|
||||||
while (await connection.Output.WaitToWriteAsync())
|
while (await connection.Output.WaitToWriteAsync())
|
||||||
{
|
{
|
||||||
if (connection.Output.TryWrite(payload))
|
if (connection.Output.TryWrite(hubMessage))
|
||||||
{
|
{
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -15,10 +15,10 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
{
|
{
|
||||||
public class HubConnectionContext
|
public class HubConnectionContext
|
||||||
{
|
{
|
||||||
private readonly WritableChannel<byte[]> _output;
|
private readonly WritableChannel<HubMessage> _output;
|
||||||
private readonly ConnectionContext _connectionContext;
|
private readonly ConnectionContext _connectionContext;
|
||||||
|
|
||||||
public HubConnectionContext(WritableChannel<byte[]> output, ConnectionContext connectionContext)
|
public HubConnectionContext(WritableChannel<HubMessage> output, ConnectionContext connectionContext)
|
||||||
{
|
{
|
||||||
_output = output;
|
_output = output;
|
||||||
_connectionContext = connectionContext;
|
_connectionContext = connectionContext;
|
||||||
|
|
@ -51,6 +51,6 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
set => DataEncoderFeature.DataEncoder = value;
|
set => DataEncoderFeature.DataEncoder = value;
|
||||||
}
|
}
|
||||||
|
|
||||||
public virtual WritableChannel<byte[]> Output => _output;
|
public virtual WritableChannel<HubMessage> Output => _output;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -53,7 +53,7 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
|
|
||||||
public async Task OnConnectedAsync(ConnectionContext connection)
|
public async Task OnConnectedAsync(ConnectionContext connection)
|
||||||
{
|
{
|
||||||
var output = Channel.CreateUnbounded<byte[]>();
|
var output = Channel.CreateUnbounded<HubMessage>();
|
||||||
|
|
||||||
// Set the hub feature before doing anything else. This stores
|
// Set the hub feature before doing anything else. This stores
|
||||||
// all the relevant state for a SignalR Hub connection.
|
// all the relevant state for a SignalR Hub connection.
|
||||||
|
|
@ -72,8 +72,9 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
{
|
{
|
||||||
while (await output.In.WaitToReadAsync())
|
while (await output.In.WaitToReadAsync())
|
||||||
{
|
{
|
||||||
while (output.In.TryRead(out var buffer))
|
while (output.In.TryRead(out var hubMessage))
|
||||||
{
|
{
|
||||||
|
var buffer = connectionContext.Protocol.WriteToArray(hubMessage);
|
||||||
buffer = encoder.Encode(buffer);
|
buffer = encoder.Encode(buffer);
|
||||||
|
|
||||||
while (await connection.Transport.Out.WaitToWriteAsync())
|
while (await connection.Transport.Out.WaitToWriteAsync())
|
||||||
|
|
@ -298,11 +299,9 @@ namespace Microsoft.AspNetCore.SignalR
|
||||||
|
|
||||||
private async Task SendMessageAsync(HubConnectionContext connection, IHubProtocol protocol, HubMessage hubMessage)
|
private async Task SendMessageAsync(HubConnectionContext connection, IHubProtocol protocol, HubMessage hubMessage)
|
||||||
{
|
{
|
||||||
var payload = protocol.WriteToArray(hubMessage);
|
|
||||||
|
|
||||||
while (await connection.Output.WaitToWriteAsync())
|
while (await connection.Output.WaitToWriteAsync())
|
||||||
{
|
{
|
||||||
if (connection.Output.TryWrite(payload))
|
if (connection.Output.TryWrite(hubMessage))
|
||||||
{
|
{
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -19,7 +19,7 @@ namespace Microsoft.AspNetCore.SignalR.Common.Protocol.Tests
|
||||||
[MemberData(nameof(HubProtocols))]
|
[MemberData(nameof(HubProtocols))]
|
||||||
public void DefaultHubProtocolResolverTestsCanCreateSupportedProtocols(IHubProtocol protocol)
|
public void DefaultHubProtocolResolverTestsCanCreateSupportedProtocols(IHubProtocol protocol)
|
||||||
{
|
{
|
||||||
var mockConnection = new Mock<HubConnectionContext>(Channel.CreateUnbounded<byte[]>().Out, new Mock<ConnectionContext>().Object);
|
var mockConnection = new Mock<HubConnectionContext>(Channel.CreateUnbounded<HubMessage>().Out, new Mock<ConnectionContext>().Object);
|
||||||
Assert.IsType(
|
Assert.IsType(
|
||||||
protocol.GetType(),
|
protocol.GetType(),
|
||||||
new DefaultHubProtocolResolver().GetProtocol(protocol.Name, mockConnection.Object));
|
new DefaultHubProtocolResolver().GetProtocol(protocol.Name, mockConnection.Object));
|
||||||
|
|
@ -30,7 +30,7 @@ namespace Microsoft.AspNetCore.SignalR.Common.Protocol.Tests
|
||||||
[InlineData("dummy")]
|
[InlineData("dummy")]
|
||||||
public void DefaultHubProtocolResolverThrowsForNotSupportedProtocol(string protocolName)
|
public void DefaultHubProtocolResolverThrowsForNotSupportedProtocol(string protocolName)
|
||||||
{
|
{
|
||||||
var mockConnection = new Mock<HubConnectionContext>(Channel.CreateUnbounded<byte[]>().Out, new Mock<ConnectionContext>().Object);
|
var mockConnection = new Mock<HubConnectionContext>(Channel.CreateUnbounded<HubMessage>().Out, new Mock<ConnectionContext>().Object);
|
||||||
var exception = Assert.Throws<NotSupportedException>(
|
var exception = Assert.Throws<NotSupportedException>(
|
||||||
() => new DefaultHubProtocolResolver().GetProtocol(protocolName, mockConnection.Object));
|
() => new DefaultHubProtocolResolver().GetProtocol(protocolName, mockConnection.Object));
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue