Remove stale .gitignore entries for moved directories

vkuttyp · Copilot · vkuttyp · commit 8e16baa9405f · 2026-03-02T14:53:16.000+03:00
PosBackend, cosmo, and cosmo-swift have been moved to CustomTests/.
Remove their now-stale ignore rules from the Swift repo .gitignore.

Co-authored-by: Copilot &lt;223556219+Copilot@users.noreply.github.com&gt;
diff --git a/.gitignore b/.gitignore
@@ -7,13 +7,6 @@ xcuserdata/
 DerivedData/
 .DS_Store
 
-# PosBackend Vapor app (separate project)
-PosBackend/
-
 # Downloaded CI logs
 logs_*/
 frpc.toml
-
-# Local scratch executables (not part of the published package)
-cosmo-swift/
-cosmo/
diff --git a/Sources/CosmoMSSQL/MSSQLConnection.swift b/Sources/CosmoMSSQL/MSSQLConnection.swift
@@ -197,7 +197,7 @@ public final class MSSQLConnection: SQLDatabase, @unchecked Sendable {
     private var isClosed:      Bool = false
     /// True when the connection is still open and usable.
     public  var isOpen:        Bool { !isClosed }
-    private var msgReader:     MessageReader?  // AsyncThrowingStream-based; no eventLoop hop per read
+    private var msgReader:     TDSFrameReader?  // AsyncThrowingStream-based; no eventLoop hop per read
     /// Tracks whether we are inside an explicit transaction (BEGIN TRANSACTION).
     private var inTransaction: Bool = false
     /// Current transaction descriptor — updated from ENVCHANGE type 8/9/10 responses.
@@ -240,15 +240,15 @@ public final class MSSQLConnection: SQLDatabase, @unchecked Sendable {
 
     private func handshake(sslContext: NIOSSLContext? = nil) async throws {
         // 1. Add pipeline: TDSTLSFramer (pass-through initially) + framing + bridge
-        let bridge = AsyncStreamBridge()
+        let bridge = TDSFrameBridge()
         // Swift 6: ByteToMessageHandler has Sendable marked unavailable (event-loop-bound).
         let bridgeBox = _UnsafeSendable(bridge)
         let frameBox  = _UnsafeSendable(ByteToMessageHandler(TDSFramingHandler()))
         let framer = tlsFramer
         try await channel.eventLoop.submit {
             try self.channel.pipeline.syncOperations.addHandlers([framer, frameBox.value, bridgeBox.value])
         }.get()
-        msgReader = MessageReader(bridge)
+        msgReader = TDSFrameReader(bridge)
 
         // 2. Pre-Login — negotiate encryption preference
         let preLoginResp = try await sendPreLogin()
@@ -428,23 +428,53 @@ public final class MSSQLConnection: SQLDatabase, @unchecked Sendable {
 
     /// Stream rows one-by-one as they are decoded from the TDS response.
     ///
-    /// The full TDS message is received before the first row is yielded (TDS framing
-    /// assembles the complete response), but the caller can process rows without
-    /// buffering the entire result set as an array.
+    /// Rows are yielded incrementally as TDS packets arrive — without buffering
+    /// the entire result set.
     public func queryStream(_ sql: String, _ binds: [SQLValue] = []) -> AsyncThrowingStream<SQLRow, Error> {
         AsyncThrowingStream { cont in
             Task { [self] in
                 do {
                     guard !self.isClosed else { throw SQLError.connectionClosed }
-                    let dec: TDSTokenDecoder
+
+                    // Send the query
                     if binds.isEmpty {
-                        dec = try await self.runBatchDecoder(sql)
+                        var payload = encodeSQLBatch(sql: sql)
+                        sendPacket(type: .sqlBatch, payload: &payload)
                     } else {
-                        dec = try await self.runRPCDecoder(Self.convertPlaceholders(sql), binds: binds)
+                        let rpc = TDSRPCRequest(sql: Self.convertPlaceholders(sql), binds: binds)
+                        var payload = rpc.encode(allocator: channel.allocator)
+                        sendPacket(type: .rpc, payload: &payload)
                     }
-                    for row in dec.rows {
-                        cont.yield(row)
+
+                    var dec = TDSTokenDecoder()
+                    var remainder: ByteBuffer? = nil
+
+                    outerLoop: while true {
+                        guard let frame = try await msgReader!.next() else {
+                            throw SQLError.connectionClosed
+                        }
+
+                        // Append this packet to any leftover bytes from the previous decode
+                        if remainder == nil || remainder!.readableBytes == 0 {
+                            remainder = frame.payload
+                        } else {
+                            var combined = channel.allocator.buffer(
+                                capacity: remainder!.readableBytes + frame.payload.readableBytes)
+                            combined.writeImmutableBuffer(remainder!)
+                            combined.writeImmutableBuffer(frame.payload)
+                            remainder = combined
+                        }
+
+                        // Decode as many complete tokens as possible
+                        let rows = dec.decodePartial(buffer: &remainder!)
+                        for row in rows { cont.yield(row) }
+
+                        if frame.isEOM { break outerLoop }
                     }
+
+                    if let err = dec.serverError { throw err }
+                    if let td = dec.transactionDescriptor { transactionDescriptor = td }
+                    dispatchInfoMessages(dec)
                     cont.finish()
                 } catch {
                     cont.finish(throwing: error)
@@ -758,10 +788,8 @@ public final class MSSQLConnection: SQLDatabase, @unchecked Sendable {
 
     /// Receive one complete TDS message via the async stream bridge handler.
     private func receivePacket() async throws -> ByteBuffer {
-        guard let reader = msgReader, let buf = try await reader.next() else {
-            throw SQLError.connectionClosed
-        }
-        return buf
+        guard let reader = msgReader else { throw SQLError.connectionClosed }
+        return try await reader.receiveMessage()
     }
 
     // MARK: - SQL Batch encoding
diff --git a/Sources/CosmoMSSQL/TDS/TDSDecoder.swift b/Sources/CosmoMSSQL/TDS/TDSDecoder.swift
@@ -55,6 +55,34 @@ struct TDSTokenDecoder {
         }
     }
 
+    /// Incrementally decode tokens from a partial buffer.
+    /// Returns rows decoded in this call. Leaves the reader index at the first
+    /// byte of any incomplete token so the caller can prepend more data and retry.
+    mutating func decodePartial(buffer: inout ByteBuffer) -> [SQLRow] {
+        var newRows: [SQLRow] = []
+        while buffer.readableBytes > 0 {
+            let savedIndex = buffer.readerIndex
+            guard let tokenByte: UInt8 = buffer.readInteger() else { break }
+            guard let token = TDSTokenType(rawValue: tokenByte) else {
+                buffer.moveReaderIndex(to: savedIndex)
+                break
+            }
+            let countBefore = currentRows.count
+            do {
+                try decodeToken(token, buffer: &buffer)
+            } catch {
+                // Incomplete — rewind to before the token byte and stop
+                buffer.moveReaderIndex(to: savedIndex)
+                break
+            }
+            let countAfter = currentRows.count
+            if countAfter > countBefore {
+                newRows.append(contentsOf: currentRows[countBefore..<countAfter])
+            }
+        }
+        return newRows
+    }
+
     // MARK: - Token dispatch
 
     private mutating func decodeToken(_ token: TDSTokenType, buffer: inout ByteBuffer) throws {
diff --git a/Sources/CosmoMSSQL/TDS/TDSHandler.swift b/Sources/CosmoMSSQL/TDS/TDSHandler.swift
@@ -3,42 +3,27 @@ import CosmoSQLCore
 
 // ── TDS NIO Channel Handler ───────────────────────────────────────────────────
 //
-// Reassembles TDS packets (which may be split across TCP segments or
-// span multiple NIO ByteBuffers) and emits complete TDS messages.
+// Emits one TDSFrame per TDS packet so consumers can process rows incrementally
+// without waiting for the entire EOM-terminated message to arrive.
 
-final class TDSFramingHandler: ByteToMessageDecoder {
-    typealias InboundOut = ByteBuffer
+/// A single TDS packet payload emitted by TDSFramingHandler.
+struct TDSFrame: @unchecked Sendable {
+    var payload: ByteBuffer
+    var isEOM:   Bool   // true if this is the last packet of the TDS message
+}
 
-    private var pendingPayload: ByteBuffer?
-    private var expectedTotal: Int = 0
-    private var isComplete: Bool = false
+final class TDSFramingHandler: ByteToMessageDecoder {
+    typealias InboundOut = TDSFrame
 
     func decode(context: ChannelHandlerContext, buffer: inout ByteBuffer) throws -> DecodingState {
-        // We need at least an 8-byte header
         guard buffer.readableBytes >= TDSPacketHeader.size else { return .needMoreData }
-
-        // Peek at header without consuming
         var peek = buffer
         let header = try TDSPacketHeader.decode(from: &peek)
-
         let packetLen = Int(header.length)
         guard buffer.readableBytes >= packetLen else { return .needMoreData }
-
-        // Consume the full packet
         var packet = buffer.readSlice(length: packetLen)!
-        packet.moveReaderIndex(forwardBy: TDSPacketHeader.size)   // skip header
-
-        if pendingPayload == nil {
-            pendingPayload = context.channel.allocator.buffer(capacity: packetLen)
-        }
-        pendingPayload!.writeBuffer(&packet)
-
-        if header.status == .eom {
-            let msg = pendingPayload!
-            pendingPayload = nil
-            context.fireChannelRead(wrapInboundOut(msg))
-        }
-
+        packet.moveReaderIndex(forwardBy: TDSPacketHeader.size)   // strip 8-byte header
+        context.fireChannelRead(wrapInboundOut(TDSFrame(payload: packet, isEOM: header.status == .eom)))
         return .continue
     }
 
@@ -49,6 +34,69 @@ final class TDSFramingHandler: ByteToMessageDecoder {
     }
 }
 
+// ── TDSFrameBridge ────────────────────────────────────────────────────────────
+//
+// Bridges per-packet TDSFrame values from the NIO pipeline into an
+// AsyncThrowingStream for consumption by async Swift callers.
+
+final class TDSFrameBridge: ChannelInboundHandler, @unchecked Sendable {
+    typealias InboundIn = TDSFrame
+
+    private let cont: AsyncThrowingStream<TDSFrame, any Error>.Continuation
+    let stream: AsyncThrowingStream<TDSFrame, any Error>
+
+    init() {
+        var captured: AsyncThrowingStream<TDSFrame, any Error>.Continuation!
+        stream = AsyncThrowingStream(bufferingPolicy: .unbounded) { captured = $0 }
+        cont = captured
+    }
+
+    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+        cont.yield(unwrapInboundIn(data))
+    }
+
+    func errorCaught(context: ChannelHandlerContext, error: any Error) {
+        cont.finish(throwing: error)
+        context.fireErrorCaught(error)
+    }
+
+    func channelInactive(context: ChannelHandlerContext) {
+        cont.finish()
+        context.fireChannelInactive()
+    }
+}
+
+// ── TDSFrameReader ────────────────────────────────────────────────────────────
+//
+// Wraps AsyncThrowingStream.AsyncIterator (a struct) in a class so it can be
+// stored and advanced from async contexts.
+
+final class TDSFrameReader: @unchecked Sendable {
+    private var iterator: AsyncThrowingStream<TDSFrame, any Error>.AsyncIterator
+
+    init(_ bridge: TDSFrameBridge) {
+        iterator = bridge.stream.makeAsyncIterator()
+    }
+
+    func next() async throws -> TDSFrame? {
+        try await iterator.next()
+    }
+
+    /// Accumulate frames until EOM — used by non-streaming callers.
+    func receiveMessage() async throws -> ByteBuffer {
+        var accumulated: ByteBuffer? = nil
+        while true {
+            guard let frame = try await next() else { throw SQLError.connectionClosed }
+            if accumulated == nil {
+                accumulated = frame.payload
+            } else {
+                accumulated!.writeImmutableBuffer(frame.payload)
+            }
+            if frame.isEOM { return accumulated! }
+        }
+    }
+}
+
 // MARK: - TDS Packet Encoder
 
 final class TDSPacketEncoder: MessageToByteEncoder {
