Add streamable multipart part #145
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,130 @@ | ||
| import HTTPTypes | ||
|
|
||
| actor StreamingMultipartPartSharedIterator< | ||
| BackingSequence: AsyncSequence, | ||
| BodyChunk: MultipartPartBodyElement | ||
| > where BackingSequence.Element == MultipartSection<BodyChunk> { | ||
| typealias BackingIterator = BackingSequence.AsyncIterator | ||
| typealias Element = StreamingMultipartPart<StreamingMultipartPartBody<BackingSequence, BodyChunk>>? | ||
|
|
||
| var pendingBodyChunk: BodyChunk? | ||
|
|
||
| private var backingIterator: BackingIterator | ||
| private var stateMachine: StateMachine | ||
| private var isReading: Bool | ||
|
|
||
| init( | ||
| makeBackingIterator: @Sendable () -> BackingIterator, | ||
| pendingBodyChunk: BodyChunk? = nil | ||
| ) { | ||
| self.backingIterator = makeBackingIterator() | ||
| self.pendingBodyChunk = pendingBodyChunk | ||
| self.stateMachine = .init() | ||
| self.isReading = false | ||
| } | ||
|
|
||
| func nextPart() async throws -> Element { | ||
| precondition(!isReading, "Streaming multipart message was iterated concurrently") | ||
| isReading = true | ||
| defer { isReading = false } | ||
|
|
||
| switch stateMachine.nextPart() { | ||
| case .currentlyStreamingBody: | ||
| throw StreamingMultipartPartError.nextPartRequestedWhileStreamingPreviousBody | ||
| case .noMoreParts: return nil | ||
| case .goodToGo: break | ||
| } | ||
|
|
||
| // if nextPartResult == .goodToGo | ||
|
|
||
| var headerFields: HTTPFields = [:] | ||
|
|
||
| while true { | ||
| nonisolated(unsafe) var iterator = backingIterator | ||
| defer { backingIterator = iterator } | ||
|
|
||
| let next: MultipartSection<BodyChunk>? | ||
| if #available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) { | ||
| next = try await iterator.next(isolation: self) | ||
| } else { | ||
| next = try await iterator.next() | ||
| } | ||
|
|
||
| guard let next else { break } | ||
|
|
||
| switch next { | ||
| case .headerFields(let fields): | ||
| headerFields.append(contentsOf: fields) | ||
| case .bodyChunk(let chunk): | ||
| let id = stateMachine.bodyStreamingStarted() | ||
| self.pendingBodyChunk = chunk | ||
| let bodySequence = StreamingMultipartPartBody(sharedIterator: self, id: id) | ||
| return StreamingMultipartPart(headerFields: headerFields, body: bodySequence) | ||
| case .boundary(let end): | ||
| if headerFields.isEmpty { | ||
| if end { | ||
| stateMachine.finish() | ||
| return nil | ||
| } | ||
| stateMachine.partStreamingEnded() | ||
| continue | ||
| } else { | ||
| // headers but no body chunk = a part with an empty body | ||
| let id = stateMachine.bodyStreamingStarted() | ||
| if end { | ||
| stateMachine.finish() | ||
| } else { | ||
| stateMachine.partStreamingEnded() | ||
| } | ||
| // state has already moved past `id`, so this body is inert | ||
| let body = StreamingMultipartPartBody(sharedIterator: self, id: id) | ||
| return StreamingMultipartPart(headerFields: headerFields, body: body) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func nextBodyChunkForSubsequence(id: Int) async throws -> BodyChunk? { | ||
| precondition(!isReading, "StreamingMultipartPart body was iterated concurrently") | ||
|
|
||
| isReading = true | ||
| defer { isReading = false } | ||
|
|
||
| switch stateMachine.nextChunk(id: id) { | ||
| case .goodToGo: break | ||
| case .endOfBody: return nil | ||
| } | ||
|
|
||
| if let pendingBodyChunk { | ||
| self.pendingBodyChunk = nil | ||
| return pendingBodyChunk | ||
| } | ||
|
|
||
| nonisolated(unsafe) var iterator = backingIterator | ||
| defer { backingIterator = iterator } | ||
|
|
||
| let next: MultipartSection<BodyChunk>? | ||
| if #available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) { | ||
| next = try await iterator.next(isolation: self) | ||
| } else { | ||
| next = try await iterator.next() | ||
| } | ||
|
|
||
| guard let next else { return nil } | ||
|
|
||
| switch next { | ||
| case .headerFields: | ||
| return nil | ||
|
Member
There was a problem hiding this comment. This would be malformed input here right? If we encounter errors during body parsing? Do we want to surface an error here instead? |
||
| case .bodyChunk(let chunk): | ||
| return chunk | ||
| case .boundary(let end): | ||
| if end { | ||
| stateMachine.finish() | ||
| } else { | ||
| stateMachine.partStreamingEnded() | ||
| } | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| import HTTPTypes | ||
|
|
||
| extension StreamingMultipartPartSharedIterator { | ||
| struct StateMachine { | ||
| enum State { | ||
| case initial | ||
| case streamingBody(id: Int) | ||
| case betweenParts | ||
| case finished | ||
| } | ||
|
|
||
| var state: State | ||
| var latestBodyID: Int = -1 | ||
|
|
||
| init() { | ||
| self.state = .initial | ||
| } | ||
|
|
||
| // Read state methods | ||
|
|
||
| enum NextPartResult { | ||
| case goodToGo | ||
| case currentlyStreamingBody | ||
| case noMoreParts | ||
| } | ||
|
|
||
| mutating func nextPart() -> NextPartResult { | ||
| switch state { | ||
| case .initial, .betweenParts: | ||
| .goodToGo | ||
| case .streamingBody: .currentlyStreamingBody | ||
| case .finished: .noMoreParts | ||
| } | ||
| } | ||
|
|
||
| enum NextChunkResult { | ||
| case goodToGo | ||
| case endOfBody | ||
| } | ||
|
|
||
| mutating func nextChunk(id: Int) -> NextChunkResult { | ||
| switch state { | ||
| case .streamingBody(let currentID) where currentID == id: .goodToGo | ||
| case .initial, .betweenParts, .finished, .streamingBody: | ||
| .endOfBody | ||
| } | ||
| } | ||
|
|
||
| // Write state methods | ||
|
|
||
| mutating func bodyStreamingStarted() -> Int { | ||
| latestBodyID += 1 | ||
| self.state = .streamingBody(id: latestBodyID) | ||
| return latestBodyID | ||
| } | ||
|
|
||
| mutating func partStreamingEnded() { | ||
| self.state = .betweenParts | ||
| } | ||
|
|
||
| mutating func finish() { | ||
| self.state = .finished | ||
| } | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| public import HTTPTypes | ||
|
|
||
| /// A single part of a multipart-encoded message whose body is streamed rather than buffered. | ||
| /// | ||
| /// This is the streaming counterpart to ``MultipartPart``: instead of holding the whole body in | ||
| /// memory, ``body`` is an `AsyncSequence` of chunks produced on demand. Parts of this kind are | ||
| /// yielded by ``StreamingMultipartPartAsyncSequence`` and expanded back into sections by | ||
| /// ``StreamingMultipartSectionAsyncSequence``, which makes them suited to large parts such as | ||
| /// file uploads. | ||
| /// | ||
| /// - Note: A part's ``body`` shares a single underlying cursor with the sequence that produced it, | ||
| /// so it must be fully consumed before the next part is requested. | ||
| public struct StreamingMultipartPart<Body: AsyncSequence & Sendable>: Sendable | ||
| where Body.Element: MultipartPartBodyElement { | ||
| /// The header fields for this part. | ||
| public let headerFields: HTTPFields | ||
|
|
||
| /// The streamed body of this part. | ||
| public let body: Body | ||
|
|
||
| /// Creates a new ``StreamingMultipartPart``. | ||
| /// | ||
| /// - Parameters: | ||
| /// - headerFields: The header fields for this part. | ||
| /// - body: The streamed body of this part. | ||
| public init(headerFields: HTTPFields, body: Body) { | ||
| self.headerFields = headerFields | ||
| self.body = body | ||
| } | ||
| } | ||
|
|
||
| /// An error thrown while consuming a ``StreamingMultipartPartAsyncSequence``. | ||
| // TODO: Make this @nonexhaustive when we drop 6.1 | ||
| public struct StreamingMultipartPartError: Error, Equatable { | ||
| enum Backing { | ||
| case nextPartRequestedWhileStreamingPreviousBody | ||
| } | ||
|
|
||
| let backing: Backing | ||
|
|
||
| init(_ backing: Backing) { | ||
| self.backing = backing | ||
| } | ||
|
|
||
| /// The next part was requested before the current part's body had been fully consumed. | ||
| /// | ||
| /// The parts and their bodies share a single underlying cursor, so each part's | ||
| /// ``StreamingMultipartPart/body`` must be fully consumed before the next part is requested. | ||
| public static let nextPartRequestedWhileStreamingPreviousBody = Self(.nextPartRequestedWhileStreamingPreviousBody) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import HTTPTypes | ||
|
|
||
| /// An asynchronous sequence that groups a stream of ``MultipartSection``s into ``StreamingMultipartPart``s. | ||
| /// | ||
| /// This sequence groups ``MultipartSection``s into parts, each made up of its header fields and a streamed | ||
| /// ``StreamingMultipartPart/body``. The body is produced on demand, so a part is never held in memory | ||
| /// in its entirety, which makes the sequence suited to large messages such as file uploads. | ||
| /// | ||
| /// ```swift | ||
| /// let parts = StreamingMultipartPartAsyncSequence(backingSequence: sections) | ||
| /// | ||
| /// for try await part in parts { | ||
| /// print(part.headerFields) | ||
| /// for try await chunk in part.body { | ||
| /// try await file.write(contentsOf: chunk) | ||
| /// } | ||
| /// } | ||
| /// ``` | ||
| /// | ||
| /// - Note: The parts and their bodies share a single underlying cursor, so each part's body must be | ||
| /// fully consumed, and parts consumed in order. Requesting the next part while a body is still | ||
| /// streaming throws ``StreamingMultipartPartError/nextPartRequestedWhileStreamingPreviousBody``. | ||
| public struct StreamingMultipartPartAsyncSequence< | ||
| BackingSequence: AsyncSequence & Sendable, | ||
| BodyChunk: MultipartPartBodyElement | ||
| >: AsyncSequence, Sendable where BackingSequence.Element == MultipartSection<BodyChunk> { | ||
| let makeBackingIterator: @Sendable () -> BackingSequence.AsyncIterator | ||
|
|
||
| /// Creates a sequence that groups the sections produced by `backingSequence` into parts. | ||
| /// | ||
| /// - Parameter backingSequence: An asynchronous sequence of ``MultipartSection``s, such as a | ||
| /// ``StreamingMultipartParserAsyncSequence``. | ||
| public init(backingSequence: BackingSequence) { | ||
| self.makeBackingIterator = { backingSequence.makeAsyncIterator() } | ||
| } | ||
|
|
||
| public struct AsyncIterator: AsyncIteratorProtocol { | ||
| public typealias Element = StreamingMultipartPart<StreamingMultipartPartBody<BackingSequence, BodyChunk>> | ||
|
|
||
| let sharedIterator: StreamingMultipartPartSharedIterator<BackingSequence, BodyChunk> | ||
|
|
||
| public func next() async throws -> Element? { | ||
| try await sharedIterator.nextPart() | ||
| } | ||
|
|
||
| @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) | ||
| public func next(isolation actor: isolated (any Actor)? = #isolation) async throws(Failure) -> Element? { | ||
| try await sharedIterator.nextPart() | ||
| } | ||
| } | ||
|
|
||
| public func makeAsyncIterator() -> AsyncIterator { | ||
| AsyncIterator(sharedIterator: .init(makeBackingIterator: makeBackingIterator)) | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,33 @@ | ||
| /// The streamed body of a ``StreamingMultipartPart``. | ||
| /// | ||
| /// Yields the part's body one ``MultipartPartBodyElement`` chunk at a time as the chunks are | ||
| /// produced by the underlying ``StreamingMultipartPartAsyncSequence``, so the whole body is never | ||
| /// held in memory at once. | ||
| /// | ||
| /// - Note: This shares a cursor with the sequence that produced the part, so it must be fully | ||
| /// consumed before the next part is requested. | ||
| public struct StreamingMultipartPartBody<BackingSequence: AsyncSequence, BodyChunk: MultipartPartBodyElement>: AsyncSequence, Sendable | ||
| where BackingSequence.Element == MultipartSection<BodyChunk> { | ||
| let sharedIterator: StreamingMultipartPartSharedIterator<BackingSequence, BodyChunk> | ||
| let id: Int | ||
|
|
||
| public struct AsyncIterator: AsyncIteratorProtocol { | ||
| public typealias Element = BodyChunk | ||
|
|
||
| let sharedIterator: StreamingMultipartPartSharedIterator<BackingSequence, BodyChunk> | ||
| let id: Int | ||
|
|
||
| public func next() async throws -> Element? { | ||
| try await sharedIterator.nextBodyChunkForSubsequence(id: id) | ||
| } | ||
|
|
||
| @available(macOS 15.0, iOS 18.0, watchOS 11.0, tvOS 18.0, visionOS 2.0, *) | ||
| public func next(isolation actor: isolated (any Actor)? = #isolation) async throws -> Element? { | ||
| try await sharedIterator.nextBodyChunkForSubsequence(id: id) | ||
| } | ||
| } | ||
|
|
||
| public func makeAsyncIterator() -> AsyncIterator { | ||
| AsyncIterator(sharedIterator: sharedIterator, id: id) | ||
| } | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Should we throw here, just in case?