Implement JobStateFetcher

avito-tech · Sep 25, 2020 · ac7e50f · ac7e50f
1 parent c34dfa8
commit ac7e50f
Show file tree

Hide file tree

Showing 25 changed files with 210 additions and 150 deletions.
diff --git a/Sources/EmceeLib/Commands/RunTestsOnRemoteQueueCommand.swift b/Sources/EmceeLib/Commands/RunTestsOnRemoteQueueCommand.swift
@@ -214,44 +214,48 @@ public final class RunTestsOnRemoteQueueCommand: Command {
             remoteCacheConfig: remoteCacheConfig,
             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)
+    }
+
+    private func waitForJobQueueToDeplete(jobId: JobId) throws {
         var caughtSignal = false
         SignalHandling.addSignalHandler(signals: [.int, .term]) { signal in
             Logger.info("Caught \(signal) signal")
-            Logger.info("Will delete job \(testArgFile.prioritizedJob.jobId)")
-            _ = try? queueClient.delete(jobId: testArgFile.prioritizedJob.jobId)
             caughtSignal = true
         }
 
-        Logger.info("Will now wait for job queue to deplete")
         try di.get(Waiter.self).waitWhile(pollPeriod: 30.0, description: "Wait for job queue to deplete") {
             if caughtSignal { return false }
-            let jobState = try queueClient.jobState(jobId: testArgFile.prioritizedJob.jobId)
-            switch jobState.queueState {
-            case .deleted:
-                return false
-            case .running(let runningQueueState):
-                BucketQueueStateLogger(runningQueueState: runningQueueState).logQueueSize()
-                return !runningQueueState.isDepleted
+
+            let state = try fetchJobState(jobId: jobId)
+            switch state.queueState {
+            case .deleted: return false
+            case .running(let queueState): return !queueState.isDepleted
             }
         }
-        Logger.info("Will now fetch job results")
-        return try queueClient.jobResults(jobId: testArgFile.prioritizedJob.jobId)
+    }
+
+    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(
+            jobId: jobId,
+            callbackQueue: callbackQueue,
+            completion: callbackWaiter.set
+        )
+        return try callbackWaiter.wait(timeout: .infinity, description: "").dematerialize()
     }
 
     private func selectPort(ports: Set<SocketModels.Port>) throws -> SocketModels.Port {
-        enum PortScanningError: Error, CustomStringConvertible {
-            case noQueuePortDetected
-
-            var description: String {
-                switch self {
-                case .noQueuePortDetected:
-                    return "No running queue server found"
-                }
-            }
+        struct NoRunningQueueFoundError: Error, CustomStringConvertible {
+            var description: String { "No running queue server found" }
         }
 
-        guard let port = ports.sorted().last else { throw PortScanningError.noQueuePortDetected }
+        guard let port = ports.sorted().last else { throw NoRunningQueueFoundError() }
         return port
     }
 }
diff --git a/Sources/QueueClient/Clients/BucketResultSender/BucketResultSenderImpl.swift b/Sources/QueueClient/Clients/BucketResultSender/BucketResultSenderImpl.swift
@@ -33,15 +33,13 @@ public final class BucketResultSenderImpl: BucketResultSender {
             request: request,
             callbackQueue: callbackQueue,
             callback: { (response: Either<BucketResultAcceptResponse, RequestSenderError>) in
-                do {
-                    let value = try response.dematerialize()
-                    switch value {
-                    case .bucketResultAccepted(let bucketId):
-                        completion(Either<BucketId, Error>.success(bucketId))
+                completion(
+                    response.mapResult {
+                        switch $0 {
+                        case .bucketResultAccepted(let bucketId): return bucketId
+                        }
                     }
-                } catch {
-                    completion(Either<BucketId, Error>.error(error))
-                }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/JobStateFetcher/JobStateFetcher.swift b/Sources/QueueClient/Clients/JobStateFetcher/JobStateFetcher.swift
@@ -0,0 +1,12 @@
+import Dispatch
+import Foundation
+import QueueModels
+import Types
+
+public protocol JobStateFetcher {
+    func fetch(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping ((Either<JobState, Error>) -> ())
+    )
+}
diff --git a/Sources/QueueClient/Clients/JobStateFetcher/JobStateFetcherImpl.swift b/Sources/QueueClient/Clients/JobStateFetcher/JobStateFetcherImpl.swift
@@ -0,0 +1,32 @@
+import Dispatch
+import Foundation
+import QueueModels
+import RESTMethods
+import RequestSender
+import Types
+
+public final class JobStateFetcherImpl: JobStateFetcher {
+    private let requestSender: RequestSender
+
+    public init(requestSender: RequestSender) {
+        self.requestSender = requestSender
+    }
+
+    public func fetch(
+        jobId: JobId,
+        callbackQueue: DispatchQueue,
+        completion: @escaping ((Either<JobState, Error>) -> ())
+    ) {
+        let request = JobStateRequest(
+            payload: JobStatePayload(
+                jobId: jobId
+            )
+        )
+        requestSender.sendRequestWithCallback(
+            request: request,
+            callbackQueue: callbackQueue
+        ) { (result: Either<JobStateResponse, RequestSenderError>) in
+            completion(result.mapResult { $0.jobState })
+        }
+    }
+}
diff --git a/Sources/QueueClient/Clients/QueueServerVersionFetcher/QueueServerVersionFetcherImpl.swift b/Sources/QueueClient/Clients/QueueServerVersionFetcher/QueueServerVersionFetcherImpl.swift
@@ -20,15 +20,13 @@ public final class QueueServerVersionFetcherImpl: QueueServerVersionFetcher {
             request: QueueVersionRequest(),
             callbackQueue: callbackQueue,
             callback: { (result: Either<QueueVersionResponse, RequestSenderError>) in
-                do {
-                    let response = try result.dematerialize()
-                    switch response {
-                    case .queueVersion(let version):
-                        completion(Either.success(version))
+                completion(
+                    result.mapResult {
+                        switch $0 {
+                        case .queueVersion(let version): return version
+                        }
                     }
-                } catch {
-                    completion(Either.error(error))
-                }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/TestScheduler/TestSchedulerImpl.swift b/Sources/QueueClient/Clients/TestScheduler/TestSchedulerImpl.swift
@@ -32,17 +32,13 @@ public final class TestSchedulerImpl: TestScheduler {
             ),
             callbackQueue: callbackQueue
         ) { (result: Either<ScheduleTestsResponse, RequestSenderError>) in
-            do {
-                let response = try result.dematerialize()
-                switch response {
-                case .scheduledTests:
-                    Logger.debug("Successfully scheduled \(testEntryConfigurations.count) tests")
-                    completion(.success(()))
+            completion(
+                result.mapResult {
+                    switch $0 {
+                    case .scheduledTests: return ()
+                    }
                 }
-            } catch {
-                Logger.error("Failed to schedule \(testEntryConfigurations.count) tests: \(error)")
-                completion(.error(error))
-            }
+            )
         }
     }
 }
diff --git a/Sources/QueueClient/Clients/WorkerDisabler/WorkerDisablerImpl.swift b/Sources/QueueClient/Clients/WorkerDisabler/WorkerDisablerImpl.swift
@@ -21,12 +21,9 @@ public final class WorkerDisablerImpl: WorkerDisabler {
             request: DisableWorkerRequest(payload: DisableWorkerPayload(workerId: workerId)),
             callbackQueue: callbackQueue,
             callback: { (result: Either<WorkerDisabledResponse, RequestSenderError>) in
-                do {
-                    let response = try result.dematerialize()
-                    completion(.success(response.workerId))
-                } catch {
-                    completion(.error(error))
-                }
+                completion(
+                    result.mapResult { $0.workerId }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/WorkerEnabler/WorkerEnablerImpl.swift b/Sources/QueueClient/Clients/WorkerEnabler/WorkerEnablerImpl.swift
@@ -21,12 +21,9 @@ public final class WorkerEnablerImpl: WorkerEnabler {
             request: EnableWorkerRequest(payload: EnableWorkerPayload(workerId: workerId)),
             callbackQueue: callbackQueue,
             callback: { (result: Either<WorkerEnabledResponse, RequestSenderError>) in
-                do {
-                    let response = try result.dematerialize()
-                    completion(.success(response.workerId))
-                } catch {
-                    completion(.error(error))
-                }
+                completion(
+                    result.mapResult { $0.workerId }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/WorkerKickstarter/WorkerKickstarterImpl.swift b/Sources/QueueClient/Clients/WorkerKickstarter/WorkerKickstarterImpl.swift
@@ -21,11 +21,9 @@ public final class WorkerKickstarterImpl: WorkerKickstarter {
             request: KickstartWorkerRequest(payload: KickstartWorkerPayload(workerId: workerId)),
             callbackQueue: callbackQueue,
             callback: { (result: Either<KickstartWorkerResponse, RequestSenderError>) in
-                do {
-                    completion(.success(try result.dematerialize().workerId))
-                } catch {
-                    completion(.error(error))
-                }
+                completion(
+                    result.mapResult { $0.workerId }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/WorkerRegisterer/WorkerRegistererImpl.swift b/Sources/QueueClient/Clients/WorkerRegisterer/WorkerRegistererImpl.swift
@@ -31,15 +31,14 @@ public final class WorkerRegistererImpl: WorkerRegisterer {
             ),
             callbackQueue: callbackQueue,
             callback: { (result: Either<RegisterWorkerResponse, RequestSenderError>) in
-                do {
-                    let response = try result.dematerialize()
-                    switch response {
-                    case .workerRegisterSuccess(let workerConfiguration):
-                        completion(Either.success(workerConfiguration))
+                completion(
+                    result.mapResult {
+                        switch $0 {
+                        case .workerRegisterSuccess(let workerConfiguration):
+                            return workerConfiguration
+                        }
                     }
-                } catch {
-                    completion(Either.error(error))
-                }
+                )
             }
         )
     }

diff --git a/Sources/QueueClient/Clients/WorkerStatus/WorkerStatusFetcherImpl.swift b/Sources/QueueClient/Clients/WorkerStatus/WorkerStatusFetcherImpl.swift
@@ -21,12 +21,9 @@ public final class WorkerStatusFetcherImpl: WorkerStatusFetcher {
             request: WorkerStatusRequest(payload: WorkerStatusPayload()),
             callbackQueue: callbackQueue
         ) { (result: Either<WorkerStatusResponse, RequestSenderError>) in
-            do {
-                let response = try result.dematerialize()
-                completion(.success(response.workerAliveness))
-            } catch {
-                completion(.error(error))
-            }
+            completion(
+                result.mapResult { $0.workerAliveness }
+            )
         }
     }
 }
diff --git a/Sources/QueueClient/QueueClient.swift b/Sources/QueueClient/QueueClient.swift
@@ -41,14 +41,6 @@ public final class QueueClient {
         )
     }
 
-    public func fetchJobState(jobId: JobId) throws {
-        try sendRequest(
-            .jobState,
-            payload: JobStateRequest(jobId: jobId),
-            completionHandler: handleJobStateResponse
-        )
-    }
-
     public func deleteJob(jobId: JobId) throws {
         try sendRequest(
             .jobDelete,
@@ -109,10 +101,6 @@ public final class QueueClient {
 
     // MARK: - Response Handlers
 
-    private func handleJobStateResponse(response: JobStateResponse) {
-        delegate?.queueClient(self, didFetchJobState: response.jobState)
-    }
-
     private func handleJobResultsResponse(response: JobResultsResponse) {
         delegate?.queueClient(self, didFetchJobResults: response.jobResults)
     }

diff --git a/Sources/QueueClient/QueueClientDelegate.swift b/Sources/QueueClient/QueueClientDelegate.swift
@@ -3,7 +3,6 @@ import QueueModels
 
 public protocol QueueClientDelegate: class {
     func queueClient(_ sender: QueueClient, didFailWithError error: QueueClientError)
-    func queueClient(_ sender: QueueClient, didFetchJobState jobState: JobState)
     func queueClient(_ sender: QueueClient, didFetchJobResults jobResults: JobResults)
     func queueClient(_ sender: QueueClient, didDeleteJob jobId: JobId)
 }
diff --git a/Sources/QueueClient/SynchronousQueueClient.swift b/Sources/QueueClient/SynchronousQueueClient.swift
@@ -12,7 +12,6 @@ import Types
 public final class SynchronousQueueClient: QueueClientDelegate {
     private let queueClient: QueueClient
     private var jobResultsResult: Either<JobResults, QueueClientError>?
-    private var jobStateResult: Either<JobState, QueueClientError>?
     private var jobDeleteResult: Either<JobId, QueueClientError>?
     private let syncQueue = DispatchQueue(label: "ru.avito.SynchronousQueueClient")
     private let requestTimeout: TimeInterval
@@ -51,19 +50,6 @@ public final class SynchronousQueueClient: QueueClientDelegate {
         }
     }
 
-    public func jobState(jobId: JobId) throws -> JobState {
-        return try synchronize {
-            jobStateResult = nil
-            return try runRetrying {
-                try queueClient.fetchJobState(jobId: jobId)
-                try SynchronousWaiter().waitWhile(timeout: requestTimeout, description: "Wait for \(jobId) job state") {
-                    self.jobStateResult == nil
-                }
-                return try jobStateResult!.dematerialize()
-            }
-        }
-    }
-
     public func delete(jobId: JobId) throws -> JobId {
         return try synchronize {
             jobDeleteResult = nil
@@ -102,13 +88,8 @@ public final class SynchronousQueueClient: QueueClientDelegate {
 
     public func queueClient(_ sender: QueueClient, didFailWithError error: QueueClientError) {
         jobResultsResult = Either.error(error)
-        jobStateResult = Either.error(error)
         jobDeleteResult = Either.error(error)
     }
-
-    public func queueClient(_ sender: QueueClient, didFetchJobState jobState: JobState) {
-        jobStateResult = Either.success(jobState)
-    }
 
     public func queueClient(_ sender: QueueClient, didFetchJobResults jobResults: JobResults) {
         jobResultsResult = Either.success(jobResults)

diff --git a/Sources/QueueServer/Endpoints/JobStateEndpoint.swift b/Sources/QueueServer/Endpoints/JobStateEndpoint.swift
@@ -6,14 +6,14 @@ import RESTServer
 
 public final class JobStateEndpoint: RESTEndpoint {
     private let stateProvider: JobStateProvider
-    public let path: RESTPath = RESTMethod.jobState
+    public let path: RESTPath = JobStateRESTMethod()
     public let requestIndicatesActivity = false
 
     public init(stateProvider: JobStateProvider) {
         self.stateProvider = stateProvider
     }
 
-    public func handle(payload: JobStateRequest) throws -> JobStateResponse {
+    public func handle(payload: JobStatePayload) throws -> JobStateResponse {
         let jobState = try stateProvider.state(jobId: payload.jobId)
         return JobStateResponse(jobState: jobState)
     }

diff --git a/Sources/RESTInterfaces/RESTMethod.swift b/Sources/RESTInterfaces/RESTMethod.swift
@@ -8,7 +8,6 @@ public enum RESTMethod: String, RESTPath {
     case getBucket
     case jobDelete
     case jobResults
-    case jobState
     case queueVersion
     case registerWorker
     case reportAlive

diff --git a/Sources/RESTMethods/JobState/JobStatePayload.swift b/Sources/RESTMethods/JobState/JobStatePayload.swift
@@ -0,0 +1,10 @@
+import Foundation
+import QueueModels
+
+public final class JobStatePayload: Codable {
+    public let jobId: JobId
+
+    public init(jobId: JobId) {
+        self.jobId = jobId
+    }
+}
diff --git a/Sources/RESTMethods/JobState/JobStateRESTMethod.swift b/Sources/RESTMethods/JobState/JobStateRESTMethod.swift
@@ -0,0 +1,8 @@
+import Foundation
+import RESTInterfaces
+
+public final class JobStateRESTMethod: RESTPath {
+    public init() {}
+
+    public var pathWithLeadingSlash: String { "/jobState" }
+}