[Runtime] SOAR-0010: Event streams sequences

Sources/OpenAPIRuntime/Multipart/ByteUtilities.swift → Sources/OpenAPIRuntime/Base/ByteUtilities.swift

@@ -24,6 +24,9 @@ enum ASCII {
     /// The line feed `<LF>` character.
     static let lf: UInt8 = 0x0a
 
+    /// The record separator `<RS>` character.
+    static let rs: UInt8 = 0x1e
+
     /// The colon `:` character.
     static let colon: UInt8 = 0x3a
 
@@ -122,3 +125,34 @@ extension RandomAccessCollection where Element: Equatable {
         return .noMatch
     }
 }
+
+/// A value returned by the `matchOfOneOf` method.
+enum MatchOfOneOfResult<C: RandomAccessCollection> {
+
+    /// No match found at any position in self.
+    case noMatch
+
+    /// The first option matched.
+    case first(C.Index)
+
+    /// The second option matched.
+    case second(C.Index)
+}
+
+extension RandomAccessCollection where Element: Equatable {
+    /// Returns the index of the first match of one of two elements.
+    /// - Parameters:
+    ///   - first: The first element to match.
+    ///   - second: The second element to match.
+    /// - Returns: The result.
+    func matchOfOneOf(first: Element, second: Element) -> MatchOfOneOfResult<Self> {
+        var index = startIndex
+        while index < endIndex {
+            let element = self[index]
+            if element == first { return .first(index) }
+            if element == second { return .second(index) }
+            formIndex(after: &index)
+        }
+        return .noMatch
+    }
+}

Sources/OpenAPIRuntime/EventStreams/JSONLinesDecoding.swift

@@ -0,0 +1,181 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftOpenAPIGenerator open source project
+//
+// Copyright (c) 2023 Apple Inc. and the SwiftOpenAPIGenerator project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftOpenAPIGenerator project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if canImport(Darwin)
+import class Foundation.JSONDecoder
+#else
+@preconcurrency import class Foundation.JSONDecoder
+#endif
+import struct Foundation.Data
+
+/// A sequence that parses arbitrary byte chunks into lines using the JSON Lines format.
+public struct JSONLinesDeserializationSequence<Upstream: AsyncSequence & Sendable>: Sendable
+where Upstream.Element == ArraySlice<UInt8> {
+
+    /// The upstream sequence.
+    private let upstream: Upstream
+
+    /// Creates a new sequence.
+    /// - Parameter upstream: The upstream sequence of arbitrary byte chunks.
+    public init(upstream: Upstream) { self.upstream = upstream }
+}
+
+extension JSONLinesDeserializationSequence: AsyncSequence {
+
+    /// The type of element produced by this asynchronous sequence.
+    public typealias Element = ArraySlice<UInt8>
+
+    /// The iterator of `JSONLinesDeserializationSequence`.
+    public struct Iterator<UpstreamIterator: AsyncIteratorProtocol>: AsyncIteratorProtocol
+    where UpstreamIterator.Element == Element {
+
+        /// The upstream iterator of arbitrary byte chunks.
+        var upstream: UpstreamIterator
+
+        /// The state machine of the iterator.
+        var stateMachine: StateMachine = .init()
+
+        /// Asynchronously advances to the next element and returns it, or ends the
+        /// sequence if there is no next element.
+        public mutating func next() async throws -> ArraySlice<UInt8>? {
+            while true {
+                switch stateMachine.next() {
+                case .returnNil: return nil
+                case .emitLine(let line): return line
+                case .needsMore:
+                    let value = try await upstream.next()
+                    switch stateMachine.receivedValue(value) {
+                    case .returnNil: return nil
+                    case .emitLine(let line): return line
+                    case .noop: continue
+                    }
+                }
+            }
+        }
+    }
+
+    /// Creates the asynchronous iterator that produces elements of this
+    /// asynchronous sequence.
+    public func makeAsyncIterator() -> Iterator<Upstream.AsyncIterator> {
+        Iterator(upstream: upstream.makeAsyncIterator())
+    }
+}
+
+extension AsyncSequence where Element == ArraySlice<UInt8> {
+
+    /// Returns another sequence that decodes each JSON Lines event as the provided type using the provided decoder.
+    /// - Parameters:
+    ///   - eventType: The type to decode the JSON event into.
+    ///   - decoder: The JSON decoder to use.
+    /// - Returns: A sequence that provides the decoded JSON events.
+    public func asDecodedJSONLines<Event: Decodable>(
+        of eventType: Event.Type = Event.self,
+        decoder: JSONDecoder = .init()
+    ) -> AsyncThrowingMapSequence<JSONLinesDeserializationSequence<Self>, Event> {
+        JSONLinesDeserializationSequence(upstream: self)
+            .map { line in try decoder.decode(Event.self, from: Data(line)) }
+    }
+}
+
+extension JSONLinesDeserializationSequence.Iterator {
+
+    /// A state machine representing the JSON Lines deserializer.
+    struct StateMachine {
+
+        /// The possible states of the state machine.
+        enum State: Hashable {
+
+            /// Is waiting for the end of line.
+            case waitingForDelimiter(buffer: [UInt8])
+
+            /// Finished, the terminal state.
+            case finished
+
+            /// Helper state to avoid copy-on-write copies.
+            case mutating
+        }
+
+        /// The current state of the state machine.
+        private(set) var state: State
+
+        /// Creates a new state machine.
+        init() { self.state = .waitingForDelimiter(buffer: []) }
+
+        /// An action returned by the `next` method.
+        enum NextAction {
+
+            /// Return nil to the caller, no more bytes.
+            case returnNil
+
+            /// Emit a full line.
+            case emitLine(ArraySlice<UInt8>)
+
+            /// The line is not complete yet, needs more bytes.
+            case needsMore
+        }
+
+        /// Read the next line parsed from upstream bytes.
+        /// - Returns: An action to perform.
+        mutating func next() -> NextAction {
+            switch state {
+            case .waitingForDelimiter(var buffer):
+                state = .mutating
+                guard let indexOfNewline = buffer.firstIndex(of: ASCII.lf) else {
+                    state = .waitingForDelimiter(buffer: buffer)
+                    return .needsMore
+                }
+                let line = buffer[..<indexOfNewline]
+                buffer.removeSubrange(...indexOfNewline)
+                state = .waitingForDelimiter(buffer: buffer)
+                return .emitLine(line)
+            case .finished: return .returnNil
+            case .mutating: preconditionFailure("Invalid state")
+            }
+        }
+
+        /// An action returned by the `receivedValue` method.
+        enum ReceivedValueAction {
+
+            /// Return nil to the caller, no more lines.
+            case returnNil
+
+            /// Emit a full line.
+            case emitLine(ArraySlice<UInt8>)
+
+            /// No action, rerun the parsing loop.
+            case noop
+        }
+
+        /// Ingest the provided bytes.
+        /// - Parameter value: A new byte chunk. If `nil`, then the source of bytes is finished.
+        /// - Returns: An action to perform.
+        mutating func receivedValue(_ value: ArraySlice<UInt8>?) -> ReceivedValueAction {
+            switch state {
+            case .waitingForDelimiter(var buffer):
+                if let value {
+                    state = .mutating
+                    buffer.append(contentsOf: value)
+                    state = .waitingForDelimiter(buffer: buffer)
+                    return .noop
+                } else {
+                    let line = ArraySlice(buffer)
+                    buffer = []
+                    state = .finished
+                    if line.isEmpty { return .returnNil } else { return .emitLine(line) }
+                }
+            case .finished, .mutating: preconditionFailure("Invalid state")
+            }
+        }
+    }
+}

Sources/OpenAPIRuntime/EventStreams/JSONLinesEncoding.swift

@@ -0,0 +1,156 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftOpenAPIGenerator open source project
+//
+// Copyright (c) 2023 Apple Inc. and the SwiftOpenAPIGenerator project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftOpenAPIGenerator project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if canImport(Darwin)
+import class Foundation.JSONEncoder
+#else
+@preconcurrency import class Foundation.JSONEncoder
+#endif
+
+/// A sequence that serializes lines by concatenating them using the JSON Lines format.
+public struct JSONLinesSerializationSequence<Upstream: AsyncSequence & Sendable>: Sendable
+where Upstream.Element == ArraySlice<UInt8> {
+
+    /// The upstream sequence.
+    private let upstream: Upstream
+
+    /// Creates a new sequence.
+    /// - Parameter upstream: The upstream sequence of lines.
+    public init(upstream: Upstream) { self.upstream = upstream }
+}
+
+extension JSONLinesSerializationSequence: AsyncSequence {
+
+    /// The type of element produced by this asynchronous sequence.
+    public typealias Element = ArraySlice<UInt8>
+
+    /// The iterator of `JSONLinesSerializationSequence`.
+    public struct Iterator<UpstreamIterator: AsyncIteratorProtocol>: AsyncIteratorProtocol
+    where UpstreamIterator.Element == Element {
+
+        /// The upstream iterator of lines.
+        var upstream: UpstreamIterator
+
+        /// The state machine of the iterator.
+        var stateMachine: StateMachine = .init()
+
+        /// Asynchronously advances to the next element and returns it, or ends the
+        /// sequence if there is no next element.
+        public mutating func next() async throws -> ArraySlice<UInt8>? {
+            while true {
+                switch stateMachine.next() {
+                case .returnNil: return nil
+                case .needsMore:
+                    let value = try await upstream.next()
+                    switch stateMachine.receivedValue(value) {
+                    case .returnNil: return nil
+                    case .emitBytes(let bytes): return bytes
+                    }
+                }
+            }
+        }
+    }
+
+    /// Creates the asynchronous iterator that produces elements of this
+    /// asynchronous sequence.
+    public func makeAsyncIterator() -> Iterator<Upstream.AsyncIterator> {
+        Iterator(upstream: upstream.makeAsyncIterator())
+    }
+}
+
+extension AsyncSequence where Element: Encodable & Sendable, Self: Sendable {
+
+    /// Returns another sequence that encodes the events using the provided encoder into JSON Lines.
+    /// - Parameter encoder: The JSON encoder to use.
+    /// - Returns: A sequence that provides the serialized JSON Lines.
+    public func asEncodedJSONLines(
+        encoder: JSONEncoder = {
+            let encoder = JSONEncoder()
+            encoder.outputFormatting = [.sortedKeys, .withoutEscapingSlashes]
+            return encoder
+        }()
+    ) -> JSONLinesSerializationSequence<AsyncThrowingMapSequence<Self, ArraySlice<UInt8>>> {
+        .init(upstream: map { event in try ArraySlice(encoder.encode(event)) })
+    }
+}
+
+extension JSONLinesSerializationSequence.Iterator {
+
+    /// A state machine representing the JSON Lines serializer.
+    struct StateMachine {
+
+        /// The possible states of the state machine.
+        enum State {
+
+            /// Is emitting serialized JSON Lines events.
+            case running
+
+            /// Finished, the terminal state.
+            case finished
+        }
+
+        /// The current state of the state machine.
+        private(set) var state: State
+
+        /// Creates a new state machine.
+        init() { self.state = .running }
+
+        /// An action returned by the `next` method.
+        enum NextAction {
+
+            /// Return nil to the caller, no more bytes.
+            case returnNil
+
+            /// Needs more bytes.
+            case needsMore
+        }
+
+        /// Read the next byte chunk serialized from upstream lines.
+        /// - Returns: An action to perform.
+        mutating func next() -> NextAction {
+            switch state {
+            case .running: return .needsMore
+            case .finished: return .returnNil
+            }
+        }
+
+        /// An action returned by the `receivedValue` method.
+        enum ReceivedValueAction {
+
+            /// Return nil to the caller, no more bytes.
+            case returnNil
+
+            /// Emit the provided bytes.
+            case emitBytes(ArraySlice<UInt8>)
+        }
+
+        /// Ingest the provided line.
+        /// - Parameter value: A new line. If `nil`, then the source of line is finished.
+        /// - Returns: An action to perform.
+        mutating func receivedValue(_ value: ArraySlice<UInt8>?) -> ReceivedValueAction {
+            switch state {
+            case .running:
+                if let value {
+                    var buffer = value
+                    buffer.append(ASCII.lf)
+                    return .emitBytes(ArraySlice(buffer))
+                } else {
+                    state = .finished
+                    return .returnNil
+                }
+            case .finished: preconditionFailure("Invalid state")
+            }
+        }
+    }
+}