Commit a4c7727e authored by Honfika's avatar Honfika

ValueTasks

parent b25d572a
...@@ -48,12 +48,12 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -48,12 +48,12 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
return reader.PeekByteFromBuffer(index); return reader.PeekByteFromBuffer(index);
} }
public Task<int> PeekByteAsync(int index, CancellationToken cancellationToken = default) public ValueTask<int> PeekByteAsync(int index, CancellationToken cancellationToken = default)
{ {
return reader.PeekByteAsync(index, cancellationToken); return reader.PeekByteAsync(index, cancellationToken);
} }
public Task<int> PeekBytesAsync(byte[] buffer, int offset, int index, int size, CancellationToken cancellationToken = default) public ValueTask<int> PeekBytesAsync(byte[] buffer, int offset, int index, int size, CancellationToken cancellationToken = default)
{ {
return reader.PeekBytesAsync(buffer, offset, index, size, cancellationToken); return reader.PeekBytesAsync(buffer, offset, index, size, cancellationToken);
} }
...@@ -124,6 +124,25 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -124,6 +124,25 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
return result; return result;
} }
public async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
int result = await reader.ReadAsync(buffer, cancellationToken);
if (result > 0)
{
if (bufferLength + result > bufferPool.BufferSize)
{
await FlushAsync(cancellationToken);
}
buffer.Span.Slice(0, result).CopyTo(new Span<byte>(this.buffer, bufferLength, result));
bufferLength += result;
ReadBytes += result;
await FlushAsync(cancellationToken);
}
return result;
}
public ValueTask<string?> ReadLineAsync(CancellationToken cancellationToken = default) public ValueTask<string?> ReadLineAsync(CancellationToken cancellationToken = default)
{ {
return CustomBufferedStream.ReadLineInternalAsync(this, bufferPool, cancellationToken); return CustomBufferedStream.ReadLineInternalAsync(this, bufferPool, cancellationToken);
......
using System.Threading; using System;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Titanium.Web.Proxy.StreamExtended.BufferPool; using Titanium.Web.Proxy.StreamExtended.BufferPool;
...@@ -93,7 +94,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -93,7 +94,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="count">The count.</param> /// <param name="count">The count.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
Task<int> ICustomStreamReader.PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken) ValueTask<int> ICustomStreamReader.PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken)
{ {
return baseStream.PeekBytesAsync(buffer, offset, index, count, cancellationToken); return baseStream.PeekBytesAsync(buffer, offset, index, count, cancellationToken);
} }
...@@ -104,7 +105,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -104,7 +105,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="index">The index.</param> /// <param name="index">The index.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
Task<int> ICustomStreamReader.PeekByteAsync(int index, CancellationToken cancellationToken) ValueTask<int> ICustomStreamReader.PeekByteAsync(int index, CancellationToken cancellationToken)
{ {
return baseStream.PeekByteAsync(index, cancellationToken); return baseStream.PeekByteAsync(index, cancellationToken);
} }
...@@ -136,6 +137,17 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -136,6 +137,17 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
return baseStream.ReadAsync(buffer, offset, count, cancellationToken); return baseStream.ReadAsync(buffer, offset, count, cancellationToken);
} }
/// <summary>
/// Reads the asynchronous.
/// </summary>
/// <param name="buffer">The buffer.</param>
/// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns>
public ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
{
return baseStream.ReadAsync(buffer, cancellationToken);
}
/// <summary> /// <summary>
/// Read a line from the byte stream /// Read a line from the byte stream
/// </summary> /// </summary>
......
...@@ -219,6 +219,45 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -219,6 +219,45 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
return available; return available;
} }
/// <summary>
/// Asynchronously reads a sequence of bytes from the current stream,
/// advances the position within the stream by the number of bytes read,
/// and monitors cancellation requests.
/// </summary>
/// <param name="buffer">The buffer to write the data into.</param>
/// <param name="cancellationToken">The token to monitor for cancellation requests.
/// The default value is <see cref="P:System.Threading.CancellationToken.None" />.</param>
/// <returns>
/// A task that represents the asynchronous read operation.
/// The value of the parameter contains the total
/// number of bytes read into the buffer.
/// The result value can be less than the number of bytes
/// requested if the number of bytes currently available is
/// less than the requested number, or it can be 0 (zero)
/// if the end of the stream has been reached.
/// </returns>
#if NETSTANDARD2_1
public override async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
#else
public async ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
#endif
{
if (bufferLength == 0)
{
await FillBufferAsync(cancellationToken);
}
int available = Math.Min(bufferLength, buffer.Length);
if (available > 0)
{
new Span<byte>(streamBuffer, bufferPos, available).CopyTo(buffer.Span);
bufferPos += available;
bufferLength -= available;
}
return available;
}
/// <summary> /// <summary>
/// Reads a byte from the stream and advances the position within the stream by one byte, or returns -1 if at the end of the stream. /// Reads a byte from the stream and advances the position within the stream by one byte, or returns -1 if at the end of the stream.
/// </summary> /// </summary>
...@@ -247,7 +286,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -247,7 +286,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="index">The index.</param> /// <param name="index">The index.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
public async Task<int> PeekByteAsync(int index, CancellationToken cancellationToken = default) public async ValueTask<int> PeekByteAsync(int index, CancellationToken cancellationToken = default)
{ {
// When index is greater than the buffer size // When index is greater than the buffer size
if (streamBuffer.Length <= index) if (streamBuffer.Length <= index)
...@@ -277,7 +316,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -277,7 +316,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="count">The count.</param> /// <param name="count">The count.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
public async Task<int> PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken = default) public async ValueTask<int> PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken = default)
{ {
// When index is greater than the buffer size // When index is greater than the buffer size
if (streamBuffer.Length <= index + count) if (streamBuffer.Length <= index + count)
......
...@@ -33,7 +33,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -33,7 +33,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="index">The index.</param> /// <param name="index">The index.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
Task<int> PeekByteAsync(int index, CancellationToken cancellationToken = default); ValueTask<int> PeekByteAsync(int index, CancellationToken cancellationToken = default);
/// <summary> /// <summary>
/// Peeks bytes asynchronous. /// Peeks bytes asynchronous.
...@@ -44,7 +44,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -44,7 +44,7 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="count">The count.</param> /// <param name="count">The count.</param>
/// <param name="cancellationToken">The cancellation token.</param> /// <param name="cancellationToken">The cancellation token.</param>
/// <returns></returns> /// <returns></returns>
Task<int> PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken = default); ValueTask<int> PeekBytesAsync(byte[] buffer, int offset, int index, int count, CancellationToken cancellationToken = default);
byte ReadByteFromBuffer(); byte ReadByteFromBuffer();
...@@ -67,8 +67,15 @@ namespace Titanium.Web.Proxy.StreamExtended.Network ...@@ -67,8 +67,15 @@ namespace Titanium.Web.Proxy.StreamExtended.Network
/// <param name="bytesToRead"></param> /// <param name="bytesToRead"></param>
/// <param name="cancellationToken"></param> /// <param name="cancellationToken"></param>
/// <returns>The number of bytes read</returns> /// <returns>The number of bytes read</returns>
Task<int> ReadAsync(byte[] buffer, int offset, int bytesToRead, Task<int> ReadAsync(byte[] buffer, int offset, int bytesToRead, CancellationToken cancellationToken = default);
CancellationToken cancellationToken = default);
/// <summary>
/// Read the specified number (or less) of raw bytes from the base stream to the given buffer to the specified offset
/// </summary>
/// <param name="buffer"></param>
/// <param name="cancellationToken"></param>
/// <returns>The number of bytes read</returns>
ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default);
/// <summary> /// <summary>
/// Read a line from the byte stream /// Read a line from the byte stream
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment