Emcee

1
import ArgLib
2
import AtomicModels
3
import EmceeDI
4
import Foundation
5
import EmceeLogging
6
import QueueClient
7
import QueueModels
8
import RequestSender
9
import SocketModels
10
import SynchronousWaiter
11
import Types
12

13
public final class EnableWorkerCommand: Command {
14
    public let name = "enableWorker"
15
    
16
    public let description = "Enables the provided worker: queue will let it execute further jobs"
17
    
18
    public var arguments: Arguments = [
19
        ArgumentDescriptions.queueServer.asRequired,
20
        ArgumentDescriptions.workerId.asRequired,
21
    ]
22
    
23
    private let callbackQueue = DispatchQueue(label: "EnableWorkerCommand.callbackQueue")
24
    private let di: DI
25
    
26
    public init(di: DI) throws {
27
        self.di = di
28
    }
29
    
30
    public func run(payload: CommandPayload) throws {
31
        let queueServerAddress: SocketAddress = try payload.expectedSingleTypedValue(argumentName: ArgumentDescriptions.queueServer.name)
32
        let workerId: WorkerId = try payload.expectedSingleTypedValue(argumentName: ArgumentDescriptions.workerId.name)
33
        
34
        let logger = try di.get(ContextualLogger.self)
35
        
36
        let workerEnabler = WorkerEnablerImpl(
37
            requestSender: try di.get(RequestSenderProvider.self).requestSender(
38
                socketAddress: queueServerAddress
39
            )
40
        )
41
        
42
        let callbackWaiter: CallbackWaiter<Either<WorkerId, Error>> = try di.get(Waiter.self).createCallbackWaiter()
43
        
44
        workerEnabler.enableWorker(
45
            workerId: workerId,
46
            callbackQueue: callbackQueue
47
        ) { (result: Either<WorkerId, Error>) in
48
            callbackWaiter.set(result: result)
49
        }
50
        
51
        let enabledWorkerId = try callbackWaiter.wait(timeout: 15, description: "Request to enable worker \(workerId) on queue")
52
        do {
53
            logger.info("Successfully enabled worker \(try enabledWorkerId.dematerialize()) on queue \(queueServerAddress)")
54
        } catch {
55
            logger.error("Failed to enable worker \(workerId) on queue \(queueServerAddress): \(error)")
56
        }
57
    }
58
}
59