diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..0973f967 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,6 @@ +.git +.build +DerivedData +Package.resolved +*.xcodeproj + diff --git a/Sources/Core/Process+Execute.swift b/Sources/Core/Process+Execute.swift index 48bb6ff0..4751556a 100644 --- a/Sources/Core/Process+Execute.swift +++ b/Sources/Core/Process+Execute.swift @@ -92,53 +92,70 @@ extension Process { /// - returns: A future containing the termination status of the process. public static func asyncExecute(_ program: String, _ arguments: [String], on worker: Worker, _ output: @escaping (ProcessOutput) -> ()) -> Future { if program.hasPrefix("/") { - let stdout = Pipe() - let stderr = Pipe() - - // will be set to false when the program is done - var running = true - - // readabilityHandler doesn't work on linux, so we are left with this hack - DispatchQueue.global().async { - while running { - let stdout = stdout.fileHandleForReading.availableData - if !stdout.isEmpty { - output(.stdout(stdout)) - } + // generic dispatch io config code for stdout/stderr + func setupDispatchIO(for pipe: Pipe, onData: @escaping (Data) -> ()) -> (Future, DispatchIO) { + // create new promise to signal IO is done + let promise = worker.eventLoop.newPromise(Void.self) + + // create dispatch io stream on read handle of pipe + let io = DispatchIO(type: .stream, fileDescriptor: pipe.fileHandleForReading.fileDescriptor, queue: executeQueue) { _ in + print("[EVENT] \(pipe.fileHandleForReading.fileDescriptor) cancel") + // close pipe once stream is cancelled + close(pipe.fileHandleForReading.fileDescriptor) } - } - DispatchQueue.global().async { - while running { - let stderr = stderr.fileHandleForReading.availableData - if !stderr.isEmpty { - output(.stderr(stderr)) + + io.setLimit(lowWater: 0) + + // start async read on stream + io.read(offset: 0, length: .max, queue: executeQueue) { done, data, err in + print("[EVENT] \(pipe.fileHandleForReading.fileDescriptor) \(done)") + if done { + // signal IO is done + promise.succeed(result: ()) + } else if err != 0 { + // signal IO failure + promise.fail(error: ProcessIOError(code: err)) + } else if let data = data, !data.isEmpty { + // convert DispatchData to Data and pass to callback + onData(Data(data)) } } + io.resume() + + // return created future and IO stream + return (promise.futureResult, io) } - - // stdout.fileHandleForReading.readabilityHandler = { handle in - // let data = handle.availableData - // guard !data.isEmpty else { - // return - // } - // output(.stdout(data)) - // } - // stderr.fileHandleForReading.readabilityHandler = { handle in - // let data = handle.availableData - // guard !data.isEmpty else { - // return - // } - // output(.stderr(data)) - // } - - let promise = worker.eventLoop.newPromise(Int32.self) - DispatchQueue.global().async { - let process = launchProcess(path: program, arguments, stdout: stdout, stderr: stderr) - process.waitUntilExit() - running = false - promise.succeed(result: process.terminationStatus) + + // create process data pipes + let stdout = Pipe() + let stderr = Pipe() + print("stdout: \(stdout.fileHandleForReading.fileDescriptor)") + print("stderr: \(stderr.fileHandleForReading.fileDescriptor)") + + // launch and run the process + let process = launchProcess(path: program, arguments, stdout: stdout, stderr: stderr) + + // setup dispatch IO on each pipe + let (stderrFuture, stderrIO) = setupDispatchIO(for: stderr) { output(.stderr($0)) } + let (stdoutFuture, stdoutIO) = setupDispatchIO(for: stdout) { output(.stdout($0)) } + + // create a new promise for the termination status and set callback + let processPromise = worker.eventLoop.newPromise(Int32.self) + process.terminationHandler = { process in + // close dispatch IO + stdoutIO.close() + stderrIO.close() + + print("process complete: \(process.terminationStatus)") + + // complete the promise + processPromise.succeed(result: process.terminationStatus) } - return promise.futureResult + process.launch() + + // combine stdout/err and process result futures + return stdoutFuture.and(stderrFuture) + .transform(to: processPromise.futureResult) } else { var resolvedPath: String? return asyncExecute("/bin/sh", ["-c", "which \(program)"], on: worker) { o in @@ -164,7 +181,6 @@ extension Process { process.arguments = arguments process.standardOutput = stdout process.standardError = stderr - process.launch() return process } } @@ -181,6 +197,10 @@ public struct ProcessExecuteError: Error { public var stdout: String } +private struct ProcessIOError: Error { + var code: Int32 +} + extension ProcessExecuteError: Debuggable { /// See `Debuggable.identifier`. public var identifier: String { @@ -193,4 +213,7 @@ extension ProcessExecuteError: Debuggable { } } +private let executeQueue = DispatchQueue(label: "codes.vapor.core.async.execute") +private let executeQueue2 = DispatchQueue(label: "codes.vapor.core.async.execute2") + #endif diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 00000000..4c828fd5 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,7 @@ +version: '2' +services: + test: + build: + context: . + dockerfile: test.Dockerfile + diff --git a/test.Dockerfile b/test.Dockerfile new file mode 100644 index 00000000..11256561 --- /dev/null +++ b/test.Dockerfile @@ -0,0 +1,4 @@ +FROM swift:4.2 +COPY . . +ENTRYPOINT swift test +