11
11
import _Concurrency
12
12
import Dispatch
13
13
import Foundation
14
+ import System
14
15
15
16
#if os(Windows)
16
17
import TSCLibc
@@ -514,14 +515,15 @@ package final class AsyncProcess {
514
515
if self . outputRedirection. redirectsOutput {
515
516
let stdoutPipe = Pipe ( )
516
517
let stderrPipe = Pipe ( )
518
+ let stdoutStream = DispatchFD ( fileHandle: stdoutPipe. fileHandleForReading) . dataStream ( )
519
+ let stderrStream = DispatchFD ( fileHandle: stderrPipe. fileHandleForReading) . dataStream ( )
517
520
518
521
group. enter ( )
519
- stdoutPipe. fileHandleForReading. readabilityHandler = { ( fh: FileHandle ) in
520
- let data = ( try ? fh. read ( upToCount: Int . max) ) ?? Data ( )
521
- if data. count == 0 {
522
- stdoutPipe. fileHandleForReading. readabilityHandler = nil
522
+ Task {
523
+ defer {
523
524
group. leave ( )
524
- } else {
525
+ }
526
+ for try await data in stdoutStream {
525
527
let contents = data. withUnsafeBytes { [ UInt8] ( $0) }
526
528
self . outputRedirection. outputClosures? . stdoutClosure ( contents)
527
529
stdoutLock. withLock {
@@ -531,16 +533,15 @@ package final class AsyncProcess {
531
533
}
532
534
533
535
group. enter ( )
534
- stderrPipe. fileHandleForReading. readabilityHandler = { ( fh: FileHandle ) in
535
- let data = ( try ? fh. read ( upToCount: Int . max) ) ?? Data ( )
536
- if data. count == 0 {
537
- stderrPipe. fileHandleForReading. readabilityHandler = nil
536
+ Task {
537
+ defer {
538
538
group. leave ( )
539
- } else {
539
+ }
540
+ for try await data in stderrStream {
540
541
let contents = data. withUnsafeBytes { [ UInt8] ( $0) }
541
- self . outputRedirection. outputClosures? . stderrClosure ( contents)
542
- stderrLock . withLock {
543
- stderr += contents
542
+ self . outputRedirection. outputClosures? . stdoutClosure ( contents)
543
+ stdoutLock . withLock {
544
+ stdout += contents
544
545
}
545
546
}
546
547
}
@@ -1354,3 +1355,54 @@ extension FileHandle: WritableByteStream {
1354
1355
}
1355
1356
}
1356
1357
#endif
1358
+
1359
+ extension DispatchFD {
1360
+ public func readChunk( upToLength maxLength: Int ) async throws -> DispatchData {
1361
+ return try await withCheckedThrowingContinuation { continuation in
1362
+ DispatchIO . read ( fromFileDescriptor: numericCast ( self . rawValue) , maxLength: maxLength, runningHandlerOn: DispatchQueue . global ( ) )
1363
+ { data, error in
1364
+ if error != 0 {
1365
+ continuation. resume ( throwing: StringError ( " POSIX error: \( error) " ) )
1366
+ return
1367
+ }
1368
+ if data. count == 0 {
1369
+ continuation. resume ( throwing: StringError ( " No more data " ) )
1370
+ }
1371
+ continuation. resume ( returning: data)
1372
+ }
1373
+ }
1374
+
1375
+ }
1376
+
1377
+ /// Returns an async stream which reads bytes from the specified file descriptor. Unlike `FileHandle.bytes`, it does not block the caller.
1378
+ @available ( macOS 15 . 0 , iOS 18 . 0 , tvOS 18 . 0 , watchOS 11 . 0 , visionOS 2 . 0 , * )
1379
+ public func dataStream( ) -> some AsyncSequence < DispatchData , any Error > {
1380
+ AsyncThrowingStream < DispatchData , any Error > {
1381
+ while !Task. isCancelled {
1382
+ let chunk = try await readChunk ( upToLength: 4096 )
1383
+ if chunk. isEmpty {
1384
+ return nil
1385
+ }
1386
+ return chunk
1387
+ }
1388
+ throw CancellationError ( )
1389
+ }
1390
+ }
1391
+ }
1392
+
1393
+ public struct DispatchFD {
1394
+ #if os(Windows)
1395
+ fileprivate let rawValue : Int
1396
+ #else
1397
+ fileprivate let rawValue : Int32
1398
+ #endif
1399
+
1400
+ init ( fileHandle: FileHandle ) {
1401
+ #if os(Windows)
1402
+ // This may look unsafe, but is how swift-corelibs-dispatch works. Basically, dispatch_fd_t directly represents either a POSIX file descriptor OR a Windows HANDLE pointer address, meaning that the fileDescriptor parameter of various Dispatch APIs is actually NOT a file descriptor on Windows but rather a HANDLE. This means that the handle should NOT be converted using _open_osfhandle, and the return value of this function should ONLY be passed to Dispatch functions where the fileDescriptor parameter is masquerading as a HANDLE in this manner. Use with extreme caution.
1403
+ rawValue = . init( bitPattern: fileHandle. _handle)
1404
+ #else
1405
+ rawValue = fileHandle. fileDescriptor
1406
+ #endif
1407
+ }
1408
+ }
0 commit comments