Add AsyncSequence helpers #1939
Conversation
fabianfett
force-pushed
the
ff-async-sequence-support
branch
from
4e590a8 to
3884bd1
Compare
fabianfett
force-pushed
the
ff-async-sequence-support
branch
from
3884bd1 to
fbdfadc
Compare
fabianfett
force-pushed
the
ff-async-sequence-support
branch
from
5613b51 to
7d09613
Compare
| public typealias AsyncIterator = Iterator | ||
|
|
||
| @usableFromInline | ||
| let upstream: Upstream |
There was a problem hiding this comment.
Nit: this may as well be var. It doesn't really matter but it can't hurt.
| /// - Parameter maxBytes: The maximum number of bytes that the result ByteBuffer is allowed to have. | ||
| /// - Returns: A ``NIO/ByteBuffer`` that holds all the bytes of the AsyncSequence | ||
| @inlinable | ||
| public func consume(maxBytes: Int) async throws -> ByteBuffer? { |
There was a problem hiding this comment.
Hmm, is returning nil here a good idea? I'm a bit inclined to want to return an empty buffer.
There was a problem hiding this comment.
consume feels to me like we're actually "reading" the bytes, collect might be more appropriate.
There was a problem hiding this comment.
I wonder if an additional lower-level variant accepting an inout ByteBuffer would also be useful? Callers could for example, reserve capacity or collect from multiple sequences into the same buffer without an intermediate copy.
There was a problem hiding this comment.
This seems like it would be more useful as a general "collect bytes from this stream of buffers" API on ByteBuffer itself.
Which means I think there might be benefit to moving it, and loosening the constraints to allow for async sequences of any buffer type (Data, Array, etc).
| switch try await upstream.next() { | ||
| case .some(let nextBuffer) where nextBuffer.readableBytes == 0: | ||
| // we received an empty buffer. for this reason, let's continue and get the | ||
| // next buffer fro, the sequence |
There was a problem hiding this comment.
| // next buffer fro, the sequence | |
| // next buffer from the sequence |
|
|
||
| case .some(var nextBuffer): | ||
| assert(nextBuffer.readableBytes > 0) | ||
| let result = nextBuffer.readInteger(as: UInt8.self) |
There was a problem hiding this comment.
Should this be banged? We assert that there must readable bytes so reading one mustn't fail
| import struct NIO.ByteBuffer | ||
|
|
||
| @available(macOS 12, iOS 15, tvOS 15, watchOS 8, *) | ||
| public struct NIOByteBufferToUInt8AsyncSequence<Upstream: AsyncSequence>: AsyncSequence where Upstream.Element == ByteBuffer { |
There was a problem hiding this comment.
I don't love the name. URLSession uses AsyncBytes (i.e. URLSession.AsyncBytes) which I think is quite clear. How about ByteBuffer.AsyncBytes?
There was a problem hiding this comment.
Is this not basically an async version of FlattenSequence/.joined() from the standard library, specialised for ByteBuffer?
So... ByteBuffer.FlattenSequence, perhaps?
| /// Consume an ``Swift/AsyncSequence`` of ``NIO/ByteBuffer``s into a single `ByteBuffer`. | ||
| /// | ||
| /// - Parameter maxBytes: The maximum number of bytes that the result ByteBuffer is allowed to have. | ||
| /// - Returns: A ``NIO/ByteBuffer`` that holds all the bytes of the AsyncSequence |
There was a problem hiding this comment.
Not necessarily all the bytes
| /// - Parameter maxBytes: The maximum number of bytes that the result ByteBuffer is allowed to have. | ||
| /// - Returns: A ``NIO/ByteBuffer`` that holds all the bytes of the AsyncSequence | ||
| @inlinable | ||
| public func consume(maxBytes: Int) async throws -> ByteBuffer? { |
There was a problem hiding this comment.
consume feels to me like we're actually "reading" the bytes, collect might be more appropriate.
| extension AsyncSequence where Element == ByteBuffer { | ||
| /// Transform an AsyncSequence of ByteBuffers into an AsyncSequence of single bytes. | ||
| @inlinable | ||
| public func toBytes() -> NIOByteBufferToUInt8AsyncSequence<Self> { |
There was a problem hiding this comment.
Should this be a computed property?
for try await byte in body.bytes {
// ...
}vs.
for try await byte in body.toBytes() {
// ...
}There was a problem hiding this comment.
joined() seems like a better name for this, to mirror both FlattenSequence and the String concatenation helper from the standard library.
| /// - Parameter maxBytes: The maximum number of bytes that the result ByteBuffer is allowed to have. | ||
| /// - Returns: A ``NIO/ByteBuffer`` that holds all the bytes of the AsyncSequence | ||
| @inlinable | ||
| public func consume(maxBytes: Int) async throws -> ByteBuffer? { |
There was a problem hiding this comment.
I wonder if an additional lower-level variant accepting an inout ByteBuffer would also be useful? Callers could for example, reserve capacity or collect from multiple sequences into the same buffer without an intermediate copy.
This PR does not have any tests yet. It is here, to demonstrate some
AsyncSequenceandByteBufferhelpers that we might want to add to make our consumers lifes' easier.Motivation:
In the [RFC] for an async/await API in AsyncHTTPClient @adam-fowler asked, if there could be a helper method to transform an AsyncSequence of
ByteBuffers into a singleByteBuffer. Following our strict rule of implementing conformances only to types you we own we must add those convenience helpers to SwiftNIO directly.Modifications:
func consume(maxBytes: Int) async throws -> ByteBufferthat consumes an AsyncSequence ofByteBuffers into a singleByteBufferfunc bytes()that translates anAsyncSequenceofByteBuffers into an AsyncSequence of Bytes (UInt8)