Code review

Author: fabianfett

Sources/_NIOConcurrency/AsyncSequenceSupport.swift

@@ -20,6 +20,19 @@ public struct ByteBufferToUInt8AsyncSequence<Upstream: AsyncSequence>: AsyncSequ
     public typealias Element = UInt8
     public typealias AsyncIterator = Iterator
 
+    @usableFromInline
+    let upstream: Upstream
+    
+    @inlinable
+    init(_ upstream: Upstream) {
+        self.upstream = upstream
+    }
+    
+    @inlinable
+    public func makeAsyncIterator() -> Iterator {
+        Iterator(self.upstream.makeAsyncIterator())
+    }
+    
     public struct Iterator: AsyncIteratorProtocol {
         /*private but*/ @usableFromInline var state: State
 
@@ -28,79 +41,66 @@ public struct ByteBufferToUInt8AsyncSequence<Upstream: AsyncSequence>: AsyncSequ
             case hasBuffer(ByteBuffer, Upstream.AsyncIterator)
             case askForMore(Upstream.AsyncIterator)
             case finished
-            case modifying
+            
+            @usableFromInline
+            init(buffer: ByteBuffer, upstream: Upstream.AsyncIterator) {
+                if buffer.readableBytes > 0 {
+                    self = .hasBuffer(buffer, upstream)
+                } else {
+                    self = .askForMore(upstream)
+                }
+            }
+            
+            @inlinable
+            mutating func next() async throws -> Element? {
+                switch self {
+                case .askForMore(var upstream):
+                    while true {
+                        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
+                            continue
+                            
+                        case .some(var nextBuffer):
+                            assert(nextBuffer.readableBytes > 0)
+                            let result = nextBuffer.readInteger(as: UInt8.self)
+                            self = .init(buffer: nextBuffer, upstream: upstream)
+                            return result
+                            
+                        case .none:
+                            self = .finished
+                            return nil
+                        }
+                    }
+                    
+                case .hasBuffer(var buffer, let upstream):
+                    assert(buffer.readableBytes > 0)
+                    let result = buffer.readInteger(as: UInt8.self)
+                    self = .init(buffer: buffer, upstream: upstream)
+                    return result
+                    
+                case .finished:
+                    return nil
+                }
+            }
         }
 
-        @usableFromInline
+        @inlinable
         init(_ upstream: Upstream.AsyncIterator) {
             self.state = .askForMore(upstream)
         }
 
         @inlinable
         public mutating func next() async throws -> Element? {
-            switch self.state {
-            case .askForMore(var upstream):
-                self.state = .modifying
-                
-                while true {
-                    switch try await upstream.next() {
-                    case .some(let nextBuffer) where nextBuffer.readableBytes == 0:
-                        break
-                        
-                    case .some(var nextBuffer):
-                        assert(nextBuffer.readableBytes > 0)
-                        let result = nextBuffer.readInteger(as: UInt8.self)
-                        if nextBuffer.readableBytes > 0 {
-                            self.state = .hasBuffer(nextBuffer, upstream)
-                        } else {
-                            self.state = .askForMore(upstream)
-                        }
-                        return result
-                        
-                    case .none:
-                        self.state = .finished
-                        return nil
-                    }
-                }
-                
-            case .hasBuffer(var buffer, let upstream):
-                assert(buffer.readableBytes > 0)
-                self.state = .modifying
-                
-                let result = buffer.readInteger(as: UInt8.self)
-                if buffer.readableBytes > 0 {
-                    self.state = .hasBuffer(buffer, upstream)
-                } else {
-                    self.state = .askForMore(upstream)
-                }
-                return result
-                
-            case .finished:
-                return nil
-                
-            case .modifying:
-                preconditionFailure("Invalid state: \(self.state)")
-            }
+            try await self.state.next()
         }
     }
 
-    @inlinable
-    public func makeAsyncIterator() -> Iterator {
-        Iterator(self.upstream.makeAsyncIterator())
-    }
-    
-    @usableFromInline
-    let upstream: Upstream
-    
-    /*private but*/ @usableFromInline init(_ upstream: Upstream) {
-        self.upstream = upstream
-    }
 }
 
-@usableFromInline
-struct TooManyBytesError: Error {
-    @usableFromInline
-    init() {}
+public struct TooManyBytesError: Error {
+    public init() {}
 }
 
 @available(macOS 12, iOS 15, tvOS 15, watchOS 8, *)