Repository: apple/swift-nio
File: Sources/NIOCore/NIODecodedAsyncSequence.swift (Swift)

Code before:
            self.baseIterator = base.asyncSequence.makeAsyncIterator()
            self.processor = NIOSingleStepByteToMessageProcessor(base.decoder)
            self.storage = Deque()
        }

        /// Retrieve the next element from the ``DecodeSequence``.
        @inlinable
        public mutating func next() async throws -> Element? {
            if !self.storage.isEmpty {
                return self.storage.removeFirst()
            }

            if self.baseIteratorIsExhausted {
                return nil
            }

            var first: Element? = nil

            loop: while first == nil {
                guard let nextBuffer = try await self.baseIterator.next() else {
                    try self.processor.finishProcessing(seenEOF: true) { decoded in
                        if first == nil {
                            first = decoded
                        } else {
                            self.storage.append(decoded)
                        }
                    }
                    self.baseIteratorIsExhausted = true
                    break loop
                }

                try self.processor.process(buffer: nextBuffer) { decoded in
                    if first == nil {
                        first = decoded
                    } else {
                        self.storage.append(decoded)
                    }
                }
            }

            /// If we've exhausted the base iterator, this _might_ be `nil`. In this case we've
            /// already set `baseIteratorIsExhausted` to true, so we're good to return `nil`.
            /// Otherwise it won't be `nil` as we'll loop until we have a value in `first`.
            return first
        }
    }
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
extension NIODecodedAsyncSequence: Sendable where AsyncSeq: Sendable, Decoder: Sendable {}


Review comment (suggestion):
Having to allocate a `Deque` is suboptimal. I didn't find functions on `NIOSingleStepByteToMessageProcessor` that would allow decoding one by one.

What should I do? Modify `NIOSingleStepByteToMessageProcessor` ? or is this good enough?
This does work fine so alternatively we can leave not-using-a-deque for another PR. 

Code after:
///     print("Decoded an element!", element)
/// }
/// ```
@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
public struct NIODecodedAsyncSequence<
    Base: AsyncSequence,
    Decoder: NIOSingleStepByteToMessageDecoder
> where Base.Element == ByteBuffer {
    @usableFromInline
    var asyncSequence: Base
    @usableFromInline
    var decoder: Decoder
    @usableFromInline
    var maximumBufferSize: Int?

    @inlinable
    init(asyncSequence: Base, decoder: Decoder, maximumBufferSize: Int? = nil) {
        self.asyncSequence = asyncSequence
        self.decoder = decoder
        self.maximumBufferSize = maximumBufferSize
    }
}

@available(macOS 10.15, iOS 13.0, watchOS 6.0, tvOS 13.0, *)
extension NIODecodedAsyncSequence: AsyncSequence {
    public typealias Element = Decoder.InboundOut

    /// Create an ``AsyncIterator`` for this ``NIODecodedAsyncSequence``.
    @inlinable
    public func makeAsyncIterator() -> AsyncIterator {
        AsyncIterator(base: self)
    }

    /// An ``AsyncIterator`` over a ``NIODecodedAsyncSequence``.
    public struct AsyncIterator: AsyncIteratorProtocol {
        @usableFromInline
        enum State: Sendable {
            case readingFromBuffer
            case readLastChunkFromBuffer
            case finishedDecoding
        }

        @usableFromInline
        var baseIterator: Base.AsyncIterator
        @usableFromInline
        var processor: NIOSingleStepByteToMessageProcessor<Decoder>
        @usableFromInline
        var state: State

        @inlinable
        init(base: NIODecodedAsyncSequence) {