JobResultFetcher

Package.swift

@@ -1137,7 +1137,6 @@ let package = Package(
                 "DistWorkerModels",
                 "Logging",
                 "QueueModels",
-                "RESTInterfaces",
                 "RESTMethods",
                 "RequestSender",
                 "ScheduleStrategy",
@@ -1153,22 +1152,15 @@ let package = Package(
             // MARK: QueueClientTests
             name: "QueueClientTests",
             dependencies: [
-                "BuildArtifactsTestHelpers",
                 "DistWorkerModels",
                 "DistWorkerModelsTestHelpers",
                 "QueueClient",
                 "QueueModels",
                 "QueueModelsTestHelpers",
-                "RESTInterfaces",
                 "RESTMethods",
                 "RequestSender",
                 "RequestSenderTestHelpers",
-                "RunnerModels",
-                "RunnerTestHelpers",
-                "SimulatorPoolTestHelpers",
                 "SocketModels",
-                "Swifter",
-                "SynchronousWaiter",
                 "TestHelpers",
                 "Types",
                 "WorkerAlivenessModels",

Sources/EmceeLib/Commands/RunTestsOnRemoteQueueCommand.swift

@@ -196,16 +196,27 @@ public final class RunTestsOnRemoteQueueCommand: Command {
             ),
             for: TestDiscoveryQuerier.self
         )
-
-        let queueClient = SynchronousQueueClient(queueServerAddress: queueServerAddress)
+        di.set(
+            JobStateFetcherImpl(
+                requestSender: try di.get(RequestSenderProvider.self).requestSender(socketAddress: queueServerAddress)
+            ),
+            for: JobStateFetcher.self
+        )
+        di.set(
+            JobResultsFetcherImpl(
+                requestSender: try di.get(RequestSenderProvider.self).requestSender(socketAddress: queueServerAddress)
+            ),
+            for: JobResultsFetcher.self
+        )
+        di.set(
+            JobDeleterImpl(
+                requestSender: try di.get(RequestSenderProvider.self).requestSender(socketAddress: queueServerAddress)
+            ),
+            for: JobDeleter.self
+        )
 
         defer {
-            Logger.info("Will delete job \(testArgFile.prioritizedJob)")
-            do {
-                _ = try queueClient.delete(jobId: testArgFile.prioritizedJob.jobId)
-            } catch {
-                Logger.error("Failed to delete job \(testArgFile.prioritizedJob): \(error)")
-            }
+            deleteJob(jobId: testArgFile.prioritizedJob.jobId)
         }
 
         try JobPreparer(di: di).formJob(
@@ -215,11 +226,8 @@ public final class RunTestsOnRemoteQueueCommand: Command {
             testArgFile: testArgFile
         )
 
-        Logger.debug("Will now wait for job queue to deplete")
         try waitForJobQueueToDeplete(jobId: testArgFile.prioritizedJob.jobId)
-
-        Logger.debug("Fetching job results")
-        return try queueClient.jobResults(jobId: testArgFile.prioritizedJob.jobId)
+        return try fetchJobResults(jobId: testArgFile.prioritizedJob.jobId)
     }
 
     private func waitForJobQueueToDeplete(jobId: JobId) throws {
@@ -229,7 +237,7 @@ public final class RunTestsOnRemoteQueueCommand: Command {
             caughtSignal = true
         }
 
-        try di.get(Waiter.self).waitWhile(pollPeriod: 30.0, description: "Wait for job queue to deplete") {
+        try di.get(Waiter.self).waitWhile(pollPeriod: 30.0, description: "Waiting for job queue to deplete") {
             if caughtSignal { return false }
 
             let state = try fetchJobState(jobId: jobId)
@@ -240,6 +248,16 @@ public final class RunTestsOnRemoteQueueCommand: Command {
         }
     }
 
+    private func fetchJobResults(jobId: JobId) throws -> JobResults {
+        let callbackWaiter: CallbackWaiter<Either<JobResults, Error>> = try di.get(Waiter.self).createCallbackWaiter()
+        try di.get(JobResultsFetcher.self).fetch(
+            jobId: jobId,
+            callbackQueue: callbackQueue,
+            completion: callbackWaiter.set
+        )
+        return try callbackWaiter.wait(timeout: .infinity, description: "Fetching job results").dematerialize()
+    }
+
     private func fetchJobState(jobId: JobId) throws -> JobState {
         let callbackWaiter: CallbackWaiter<Either<JobState, Error>> =  try di.get(Waiter.self).createCallbackWaiter()
         try di.get(JobStateFetcher.self).fetch(
@@ -258,4 +276,18 @@ public final class RunTestsOnRemoteQueueCommand: Command {
         guard let port = ports.sorted().last else { throw NoRunningQueueFoundError() }
         return port
     }
+
+    private func deleteJob(jobId: JobId) {
+        do {
+            let callbackWaiter: CallbackWaiter<Either<(), Error>> = try di.get(Waiter.self).createCallbackWaiter()
+            try di.get(JobDeleter.self).delete(
+                jobId: jobId,
+                callbackQueue: callbackQueue,
+                completion: callbackWaiter.set
+            )
+            try callbackWaiter.wait(timeout: .infinity, description: "Deleting job").dematerialize()
+        } catch {
+            Logger.error("Failed to delete job: \(error)")
+        }
+    }
 }

Sources/QueueClient/Clients/JobDeleter/JobDeleter.swift

@@ -0,0 +1,12 @@
+import Dispatch
+import Foundation
+import QueueModels
+import Types
+
+public protocol JobDeleter {
+    func delete(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping (Either<(), Error>) -> ()
+    )
+}

Sources/QueueClient/Clients/JobDeleter/JobDeleterImpl.swift

@@ -0,0 +1,32 @@
+import Dispatch
+import Foundation
+import QueueModels
+import RESTMethods
+import RequestSender
+import Types
+
+public final class JobDeleterImpl: JobDeleter {
+    private let requestSender: RequestSender
+
+    public init(requestSender: RequestSender) {
+        self.requestSender = requestSender
+    }
+
+    public func delete(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping (Either<(), Error>) -> ()
+    ) {
+        let request = JobDeleteRequest(
+            payload: JobDeletePayload(
+                jobId: jobId
+            )
+        )
+        requestSender.sendRequestWithCallback(
+            request: request,
+            callbackQueue: callbackQueue
+        ) { (result: Either<JobDeleteResponse, RequestSenderError>) in
+            completion(result.mapResult { _ in () })
+        }
+    }
+}

Sources/QueueClient/Clients/JobResultsFetcher/JobResultsFetcher.swift

@@ -0,0 +1,12 @@
+import Dispatch
+import Foundation
+import QueueModels
+import Types
+
+public protocol JobResultsFetcher {
+    func fetch(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping (Either<JobResults, Error>) -> ()
+    )
+}

Sources/QueueClient/Clients/JobResultsFetcher/JobResultsFetcherImpl.swift

@@ -0,0 +1,35 @@
+import Foundation
+import QueueModels
+import RESTMethods
+import RequestSender
+import Types
+
+public final class JobResultsFetcherImpl: JobResultsFetcher {
+    private let requestSender: RequestSender
+
+    public init(requestSender: RequestSender) {
+        self.requestSender = requestSender
+    }
+
+    public func fetch(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping (Either<JobResults, Error>) -> ()
+    ) {
+        let request = JobResultRequest(
+            payload: JobResultsPayload(
+                jobId: jobId
+            )
+        )
+
+        requestSender.sendRequestWithCallback(
+            request: request,
+            callbackQueue: callbackQueue,
+            callback: { (response: Either<JobResultsResponse, RequestSenderError>) in
+                completion(
+                    response.mapResult { $0.jobResults }
+                )
+            }
+        )
+    }
+}