huahaofeng
/
aspnetcore
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

React to pipeline changes (#1365)

This commit is contained in:
Pavel Krymets 2018-01-29 14:36:03 -08:00 committed by GitHub
parent 6590992c96
commit ac9db5dc7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 46 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
build/dependencies.props
View File

@ -59,14 +59,14 @@
<MsgPackCliPackageVersion>0.9.0-beta2</MsgPackCliPackageVersion> <MsgPackCliPackageVersion>0.9.0-beta2</MsgPackCliPackageVersion>
<NewtonsoftJsonPackageVersion>10.0.1</NewtonsoftJsonPackageVersion> <NewtonsoftJsonPackageVersion>10.0.1</NewtonsoftJsonPackageVersion>
<StackExchangeRedisStrongNamePackageVersion>1.2.4</StackExchangeRedisStrongNamePackageVersion> <StackExchangeRedisStrongNamePackageVersion>1.2.4</StackExchangeRedisStrongNamePackageVersion>
<SystemBuffersPackageVersion>4.5.0-preview1-26112-01</SystemBuffersPackageVersion> <SystemBuffersPackageVersion>4.5.0-preview1-26126-05</SystemBuffersPackageVersion>
<SystemIOPipelinesPackageVersion>0.1.0-e180104-2</SystemIOPipelinesPackageVersion> <SystemIOPipelinesPackageVersion>0.1.0-preview1-180129-1</SystemIOPipelinesPackageVersion>
<SystemMemoryPackageVersion>4.5.0-preview1-26112-01</SystemMemoryPackageVersion> <SystemMemoryPackageVersion>4.5.0-preview1-26126-05</SystemMemoryPackageVersion>
<SystemNumericsVectorsPackageVersion>4.5.0-preview1-26112-01</SystemNumericsVectorsPackageVersion> <SystemNumericsVectorsPackageVersion>4.5.0-preview1-26126-05</SystemNumericsVectorsPackageVersion>
<SystemReactiveLinqPackageVersion>3.1.1</SystemReactiveLinqPackageVersion> <SystemReactiveLinqPackageVersion>3.1.1</SystemReactiveLinqPackageVersion>
<SystemReflectionEmitPackageVersion>4.3.0</SystemReflectionEmitPackageVersion> <SystemReflectionEmitPackageVersion>4.3.0</SystemReflectionEmitPackageVersion>
<SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview1-26112-01</SystemRuntimeCompilerServicesUnsafePackageVersion> <SystemRuntimeCompilerServicesUnsafePackageVersion>4.5.0-preview1-26126-05</SystemRuntimeCompilerServicesUnsafePackageVersion>
<SystemThreadingChannelsPackageVersion>4.5.0-preview1-26112-01</SystemThreadingChannelsPackageVersion> <SystemThreadingChannelsPackageVersion>4.5.0-preview1-26126-05</SystemThreadingChannelsPackageVersion>
<SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview1-26112-01</SystemThreadingTasksExtensionsPackageVersion> <SystemThreadingTasksExtensionsPackageVersion>4.5.0-preview1-26112-01</SystemThreadingTasksExtensionsPackageVersion>
<XunitPackageVersion>2.3.1</XunitPackageVersion> <XunitPackageVersion>2.3.1</XunitPackageVersion>
<XunitRunnerVisualStudioPackageVersion>2.3.1</XunitRunnerVisualStudioPackageVersion> <XunitRunnerVisualStudioPackageVersion>2.3.1</XunitRunnerVisualStudioPackageVersion>

4
src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/PipelineReaderExtensions.cs
View File

@ -8,7 +8,7 @@ namespace System.IO.Pipelines
{ {
internal static class PipelineReaderExtensions internal static class PipelineReaderExtensions
{ {
public static async Task CopyToAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) public static async Task CopyToAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
{ {
// TODO: Use bufferSize argument // TODO: Use bufferSize argument
while (!cancellationToken.IsCancellationRequested) while (!cancellationToken.IsCancellationRequested)
@ -26,7 +26,7 @@ namespace System.IO.Pipelines
} }
finally finally
{ {
input.Advance(inputBuffer.End); input.AdvanceTo(inputBuffer.End);
} }
} }
} }

29
src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/StreamExtensions.cs
View File

@ -11,19 +11,19 @@ namespace System.IO.Pipelines
internal static class StreamExtensions internal static class StreamExtensions
{ {
/// <summary> /// <summary>
/// Copies the content of a <see cref="Stream"/> into a <see cref="IPipeWriter"/>. /// Copies the content of a <see cref="Stream"/> into a <see cref="PipeWriter"/>.
/// </summary> /// </summary>
/// <param name="stream"></param> /// <param name="stream"></param>
/// <param name="writer"></param> /// <param name="writer"></param>
/// <param name="cancellationToken"></param> /// <param name="cancellationToken"></param>
/// <returns></returns> /// <returns></returns>
public static Task CopyToAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default) public static Task CopyToAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default)
{ {
// 81920 is the default bufferSize, there is not stream.CopyToAsync overload that takes only a cancellationToken // 81920 is the default bufferSize, there is not stream.CopyToAsync overload that takes only a cancellationToken
return stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken); return stream.CopyToAsync(new PipelineWriterStream(writer), bufferSize: 81920, cancellationToken: cancellationToken);
} }
public static async Task CopyToEndAsync(this Stream stream, IPipeWriter writer, CancellationToken cancellationToken = default) public static async Task CopyToEndAsync(this Stream stream, PipeWriter writer, CancellationToken cancellationToken = default)
{ {
try try
{ {
@ -38,14 +38,14 @@ namespace System.IO.Pipelines
} }
/// <summary> /// <summary>
/// Copies a <see cref="ReadOnlyBuffer"/> to a <see cref="Stream"/> asynchronously /// Copies a <see cref="ReadOnlyBuffer{Byte}"/> to a <see cref="Stream"/> asynchronously
/// </summary> /// </summary>
/// <param name="buffer">The <see cref="ReadOnlyBuffer"/> to copy</param> /// <param name="buffer">The <see cref="ReadOnlyBuffer{Byte}"/> to copy</param>
/// <param name="stream">The target <see cref="Stream"/></param> /// <param name="stream">The target <see cref="Stream"/></param>
/// <returns></returns> /// <returns></returns>
public static Task CopyToAsync(this ReadOnlyBuffer buffer, Stream stream) public static Task CopyToAsync(this ReadOnlyBuffer<byte> buffer, Stream stream)
{ {
if (buffer.IsSingleSpan) if (buffer.IsSingleSegment)
{ {
return WriteToStream(stream, buffer.First); return WriteToStream(stream, buffer.First);
} }
@ -53,7 +53,7 @@ namespace System.IO.Pipelines
return CopyMultipleToStreamAsync(buffer, stream); return CopyMultipleToStreamAsync(buffer, stream);
} }
private static async Task CopyMultipleToStreamAsync(this ReadOnlyBuffer buffer, Stream stream) private static async Task CopyMultipleToStreamAsync(this ReadOnlyBuffer<byte> buffer, Stream stream)
{ {
foreach (var memory in buffer) foreach (var memory in buffer)
{ {
@ -77,12 +77,12 @@ namespace System.IO.Pipelines
} }
} }
public static Task CopyToEndAsync(this IPipeReader input, Stream stream) public static Task CopyToEndAsync(this PipeReader input, Stream stream)
{ {
return input.CopyToEndAsync(stream, 4096, CancellationToken.None); return input.CopyToEndAsync(stream, 4096, CancellationToken.None);
} }
public static async Task CopyToEndAsync(this IPipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken) public static async Task CopyToEndAsync(this PipeReader input, Stream stream, int bufferSize, CancellationToken cancellationToken)
{ {
try try
{ {
@ -98,9 +98,9 @@ namespace System.IO.Pipelines
private class PipelineWriterStream : Stream private class PipelineWriterStream : Stream
{ {
private readonly IPipeWriter _writer; private readonly PipeWriter _writer;
public PipelineWriterStream(IPipeWriter writer) public PipelineWriterStream(PipeWriter writer)
{ {
_writer = writer; _writer = writer;
} }
@ -148,9 +148,8 @@ namespace System.IO.Pipelines
{ {
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
var output = _writer.Alloc(); _writer.Write(new ReadOnlySpan<byte>(buffer, offset, count));
output.Write(new ReadOnlySpan<byte>(buffer, offset, count)); await _writer.FlushAsync(cancellationToken);
await output.FlushAsync(cancellationToken);
} }
} }
} }

10
src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/StreamPipeConnection.cs
View File

@ -3,7 +3,7 @@
namespace System.IO.Pipelines namespace System.IO.Pipelines
{ {
internal class StreamPipeConnection : IPipeConnection internal class StreamPipeConnection : IDuplexPipe
{ {
public StreamPipeConnection(PipeOptions options, Stream stream) public StreamPipeConnection(PipeOptions options, Stream stream)
{ {
@ -11,9 +11,9 @@ namespace System.IO.Pipelines
Output = CreateWriter(options, stream); Output = CreateWriter(options, stream);
} }
public IPipeReader Input { get; } public PipeReader Input { get; }
public IPipeWriter Output { get; } public PipeWriter Output { get; }
public void Dispose() public void Dispose()
{ {
@ -21,7 +21,7 @@ namespace System.IO.Pipelines
Output.Complete(); Output.Complete();
} }
public static IPipeReader CreateReader(PipeOptions options, Stream stream) public static PipeReader CreateReader(PipeOptions options, Stream stream)
{ {
if (!stream.CanRead) if (!stream.CanRead)
{ {
@ -34,7 +34,7 @@ namespace System.IO.Pipelines
return pipe.Reader; return pipe.Reader;
} }
public static IPipeWriter CreateWriter(PipeOptions options, Stream stream) public static PipeWriter CreateWriter(PipeOptions options, Stream stream)
{ {
if (!stream.CanWrite) if (!stream.CanWrite)
{ {

11
src/Microsoft.AspNetCore.Sockets.Client.Http/ServerSentEventsMessageParser.cs
View File

@ -3,6 +3,7 @@
using System; using System;
using System.Buffers; using System.Buffers;
using System.Collections;
using System.Collections.Generic; using System.Collections.Generic;
using System.Collections.Sequences; using System.Collections.Sequences;
using System.IO.Pipelines; using System.IO.Pipelines;
@ -24,7 +25,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
private InternalParseState _internalParserState = InternalParseState.ReadMessagePayload; private InternalParseState _internalParserState = InternalParseState.ReadMessagePayload;
private List<byte[]> _data = new List<byte[]>(); private List<byte[]> _data = new List<byte[]>();
public ParseResult ParseMessage(ReadOnlyBuffer buffer, out Position consumed, out Position examined, out byte[] message) public ParseResult ParseMessage(ReadOnlyBuffer<byte> buffer, out SequencePosition consumed, out SequencePosition examined, out byte[] message)
{ {
consumed = buffer.Start; consumed = buffer.Start;
examined = buffer.End; examined = buffer.End;
@ -35,7 +36,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
while (buffer.Length > 0) while (buffer.Length > 0)
{ {
if (ReadOnlyBuffer.Seek(start, end, out var lineEnd, ByteLF) == -1) if (!(buffer.PositionOf(ByteLF) is SequencePosition lineEnd))
{ {
// For the case of data: Foo\r\n\r\<Anytine except \n> // For the case of data: Foo\r\n\r\<Anytine except \n>
if (_internalParserState == InternalParseState.ReadEndOfMessage) if (_internalParserState == InternalParseState.ReadEndOfMessage)
@ -50,7 +51,7 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
return ParseResult.Incomplete; return ParseResult.Incomplete;
} }
lineEnd = buffer.Move(lineEnd, 1); lineEnd = buffer.GetPosition(lineEnd, 1);
var line = ConvertBufferToSpan(buffer.Slice(start, lineEnd)); var line = ConvertBufferToSpan(buffer.Slice(start, lineEnd));
buffer = buffer.Slice(line.Length); buffer = buffer.Slice(line.Length);
@ -148,9 +149,9 @@ namespace Microsoft.AspNetCore.Sockets.Internal.Formatters
} }
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private ReadOnlySpan<byte> ConvertBufferToSpan(ReadOnlyBuffer buffer) private ReadOnlySpan<byte> ConvertBufferToSpan(ReadOnlyBuffer<byte> buffer)
{ {
if (buffer.IsSingleSpan) if (buffer.IsSingleSegment)
{ {
return buffer.First.Span; return buffer.First.Span;
} }

4
src/Microsoft.AspNetCore.Sockets.Client.Http/ServerSentEventsTransport.cs
View File

@ -86,7 +86,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
var stream = await response.Content.ReadAsStreamAsync(); var stream = await response.Content.ReadAsStreamAsync();
var pipelineReader = StreamPipeConnection.CreateReader(new PipeOptions(_memoryPool), stream); var pipelineReader = StreamPipeConnection.CreateReader(new PipeOptions(_memoryPool), stream);
var readCancellationRegistration = cancellationToken.Register( var readCancellationRegistration = cancellationToken.Register(
reader => ((IPipeReader)reader).CancelPendingRead(), pipelineReader); reader => ((PipeReader)reader).CancelPendingRead(), pipelineReader);
try try
{ {
while (true) while (true)
@ -122,7 +122,7 @@ namespace Microsoft.AspNetCore.Sockets.Client
} }
finally finally
{ {
pipelineReader.Advance(consumed, examined); pipelineReader.AdvanceTo(consumed, examined);
} }
} }
} }

21
test/Microsoft.AspNetCore.SignalR.Client.Tests/ServerSentEventsParserTests.cs
View File

@ -3,6 +3,7 @@
using System; using System;
using System.Buffers; using System.Buffers;
using System.Collections;
using System.Collections.Generic; using System.Collections.Generic;
using System.Collections.Sequences; using System.Collections.Sequences;
using System.IO.Pipelines; using System.IO.Pipelines;
@ -32,7 +33,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
public void ParseSSEMessageSuccessCases(string encodedMessage, string expectedMessage) public void ParseSSEMessageSuccessCases(string encodedMessage, string expectedMessage)
{ {
var buffer = Encoding.UTF8.GetBytes(encodedMessage); var buffer = Encoding.UTF8.GetBytes(encodedMessage);
var readableBuffer = new ReadOnlyBuffer(buffer); var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
var parser = new ServerSentEventsMessageParser(); var parser = new ServerSentEventsMessageParser();
var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);
@ -58,7 +59,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
public void ParseSSEMessageFailureCases(string encodedMessage, string expectedExceptionMessage) public void ParseSSEMessageFailureCases(string encodedMessage, string expectedExceptionMessage)
{ {
var buffer = Encoding.UTF8.GetBytes(encodedMessage); var buffer = Encoding.UTF8.GetBytes(encodedMessage);
var readableBuffer = new ReadOnlyBuffer(buffer); var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
var parser = new ServerSentEventsMessageParser(); var parser = new ServerSentEventsMessageParser();
var ex = Assert.Throws<FormatException>(() => { parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); }); var ex = Assert.Throws<FormatException>(() => { parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); });
@ -86,7 +87,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
public void ParseSSEMessageIncompleteParseResult(string encodedMessage) public void ParseSSEMessageIncompleteParseResult(string encodedMessage)
{ {
var buffer = Encoding.UTF8.GetBytes(encodedMessage); var buffer = Encoding.UTF8.GetBytes(encodedMessage);
var readableBuffer = new ReadOnlyBuffer(buffer); var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
var parser = new ServerSentEventsMessageParser(); var parser = new ServerSentEventsMessageParser();
var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);
@ -114,7 +115,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
var pipe = new Pipe(new PipeOptions(pool)); var pipe = new Pipe(new PipeOptions(pool));
byte[] message = null; byte[] message = null;
Position consumed = default, examined = default; SequencePosition consumed = default, examined = default;
for (var i = 0; i < messageParts.Length; i++) for (var i = 0; i < messageParts.Length; i++)
{ {
@ -123,7 +124,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
var result = await pipe.Reader.ReadAsync(); var result = await pipe.Reader.ReadAsync();
var parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message); var parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message);
pipe.Reader.Advance(consumed, examined); pipe.Reader.AdvanceTo(consumed, examined);
// parse result should be complete only after we parsed the last message part // parse result should be complete only after we parsed the last message part
var expectedResult = var expectedResult =
@ -167,7 +168,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out var buffer); var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out var buffer);
Assert.Equal(ServerSentEventsMessageParser.ParseResult.Incomplete, parseResult); Assert.Equal(ServerSentEventsMessageParser.ParseResult.Incomplete, parseResult);
pipe.Reader.Advance(consumed, examined); pipe.Reader.AdvanceTo(consumed, examined);
// Send the rest of the data and parse the complete message // Send the rest of the data and parse the complete message
await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart2)); await pipe.Writer.WriteAsync(Encoding.UTF8.GetBytes(encodedMessagePart2));
@ -195,8 +196,8 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out var message); var parseResult = parser.ParseMessage(result.Buffer, out var consumed, out var examined, out var message);
Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult);
Assert.Equal("foo", Encoding.UTF8.GetString(message)); Assert.Equal("foo", Encoding.UTF8.GetString(message));
Assert.Equal(consumed, result.Buffer.Move(result.Buffer.Start, message1.Length)); Assert.Equal(consumed, result.Buffer.GetPosition(result.Buffer.Start, message1.Length));
pipe.Reader.Advance(consumed, examined); pipe.Reader.AdvanceTo(consumed, examined);
Assert.Equal(consumed, examined); Assert.Equal(consumed, examined);
parser.Reset(); parser.Reset();
@ -205,7 +206,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message); parseResult = parser.ParseMessage(result.Buffer, out consumed, out examined, out message);
Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult); Assert.Equal(ServerSentEventsMessageParser.ParseResult.Completed, parseResult);
Assert.Equal("bar", Encoding.UTF8.GetString(message)); Assert.Equal("bar", Encoding.UTF8.GetString(message));
pipe.Reader.Advance(consumed, examined); pipe.Reader.AdvanceTo(consumed, examined);
} }
} }
@ -223,7 +224,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests
public void ParseMessagesWithMultipleDataLines(string encodedMessage, string expectedMessage) public void ParseMessagesWithMultipleDataLines(string encodedMessage, string expectedMessage)
{ {
var buffer = Encoding.UTF8.GetBytes(encodedMessage); var buffer = Encoding.UTF8.GetBytes(encodedMessage);
var readableBuffer = new ReadOnlyBuffer(buffer); var readableBuffer = new ReadOnlyBuffer<byte>(buffer);
var parser = new ServerSentEventsMessageParser(); var parser = new ServerSentEventsMessageParser();
var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message); var parseResult = parser.ParseMessage(readableBuffer, out var consumed, out var examined, out var message);