Emcee

1
import AutomaticTermination
2
import Deployer
3
import DistWorkerModels
4
import EmceeExtensions
5
import Foundation
6
import MetricsExtensions
7
import QueueModels
8

9
public struct QueueServerConfiguration: Codable {
10
    
11
    /// Global analytics. https://github.com/avito-tech/Emcee/releases/tag/12.0.0
12
    public let globalAnalyticsConfiguration: AnalyticsConfiguration
13
    
14
    /// How often workers should poll a queue for new buckets, in seconds. Recommended and default value is `30` seconds.
15
    public let checkAgainTimeInterval: TimeInterval
16
    
17
    /// Where queue should be deployed.
18
    public let queueServerDeploymentDestinations: [DeploymentDestination]
19
    
20
    /// How queue should terminate.
21
    public let queueServerTerminationPolicy: AutomaticTerminationPolicy
22
    
23
    /// Where workers expected to be started.
24
    public let workerDeploymentDestinations: [DeploymentDestination]
25
    
26
    /// Default worker configuration, in case if you don't specify ones in `workerDeploymentDestinations`.
27
    public let defaultWorkerConfiguration: WorkerSpecificConfiguration?
28
    
29
    /// How workers are started.
30
    public let workerStartMode: WorkerStartMode
31
    
32
    /// Force communication only over IPv4. Sometimes IPv6 is enabled but not configured properly. In most cases pass `true` for ease of use.
33
    public let useOnlyIPv4: Bool
34
    
35
    /// What ports Emcee queue and workers should use. Default is `41000 ... 41010`.
36
    public let portRange: PortRange
37
    
38
    public init(
39
        globalAnalyticsConfiguration: AnalyticsConfiguration,
40
        checkAgainTimeInterval: TimeInterval,
41
        queueServerDeploymentDestinations: [DeploymentDestination],
42
        queueServerTerminationPolicy: AutomaticTerminationPolicy,
43
        workerDeploymentDestinations: [DeploymentDestination],
44
        defaultWorkerSpecificConfiguration: WorkerSpecificConfiguration?,
45
        workerStartMode: WorkerStartMode,
46
        useOnlyIPv4: Bool,
47
        portRange: PortRange
48
    ) {
49
        self.globalAnalyticsConfiguration = globalAnalyticsConfiguration
50
        self.checkAgainTimeInterval = checkAgainTimeInterval
51
        self.queueServerDeploymentDestinations = queueServerDeploymentDestinations
52
        self.queueServerTerminationPolicy = queueServerTerminationPolicy
53
        self.workerDeploymentDestinations = workerDeploymentDestinations
54
        self.defaultWorkerConfiguration = defaultWorkerSpecificConfiguration
55
        self.workerStartMode = workerStartMode
56
        self.useOnlyIPv4 = useOnlyIPv4
57
        self.portRange = portRange
58
    }
59
    
60
    private enum CodingKeys: String, CodingKey {
61
        case globalAnalyticsConfiguration
62
        case checkAgainTimeInterval
63
        case queueServerDeploymentDestinations
64
        case queueServerTerminationPolicy
65
        case workerDeploymentDestinations
66
        case defaultWorkerConfiguration
67
        case workerStartMode
68
        case useOnlyIPv4
69
        case portRange
70
    }
71
    
72
    public init(from decoder: Decoder) throws {
73
        let container = try decoder.container(keyedBy: CodingKeys.self)
74
        
75
        let globalAnalyticsConfiguration = try container.decodeIfPresentExplaining(AnalyticsConfiguration.self, forKey: .globalAnalyticsConfiguration) ?? QueueServerConfigurationDefaultValues.globalAnalyticsConfiguration
76
        let checkAgainTimeInterval = try container.decodeIfPresentExplaining(TimeInterval.self, forKey: .checkAgainTimeInterval) ?? QueueServerConfigurationDefaultValues.checkAgainTimeInterval
77
        let queueServerDeploymentDestinations = try container.decodeExplaining([DeploymentDestination].self, forKey: .queueServerDeploymentDestinations)
78
        let queueServerTerminationPolicy = try container.decodeIfPresentExplaining(AutomaticTerminationPolicy.self, forKey: .queueServerTerminationPolicy) ?? QueueServerConfigurationDefaultValues.queueServerTerminationPolicy
79
        let workerDeploymentDestinations = try container.decodeExplaining([DeploymentDestination].self, forKey: .workerDeploymentDestinations)
80
        let defaultWorkerSpecificConfiguration = try container.decodeIfPresentExplaining(WorkerSpecificConfiguration.self, forKey: .defaultWorkerConfiguration) ?? WorkerSpecificConfigurationDefaultValues.defaultWorkerConfiguration
81
        let workerStartMode = try container.decodeIfPresentExplaining(WorkerStartMode.self, forKey: .workerStartMode) ?? QueueServerConfigurationDefaultValues.workerStartMode
82
        let useOnlyIPv4 = try container.decodeIfPresentExplaining(Bool.self, forKey: .useOnlyIPv4) ?? QueueServerConfigurationDefaultValues.useOnlyIPv4
83
        let portRange = try container.decodeIfPresentExplaining(PortRange.self, forKey: .portRange) ?? QueueServerConfigurationDefaultValues.defaultQueuePortRange
84
        
85
        self.init(
86
            globalAnalyticsConfiguration: globalAnalyticsConfiguration,
87
            checkAgainTimeInterval: checkAgainTimeInterval,
88
            queueServerDeploymentDestinations: queueServerDeploymentDestinations,
89
            queueServerTerminationPolicy: queueServerTerminationPolicy,
90
            workerDeploymentDestinations: workerDeploymentDestinations,
91
            defaultWorkerSpecificConfiguration: defaultWorkerSpecificConfiguration,
92
            workerStartMode: workerStartMode,
93
            useOnlyIPv4: useOnlyIPv4,
94
            portRange: portRange
95
        )
96
    }
97
    
98
    public func encode(to encoder: Encoder) throws {
99
        var container = encoder.container(keyedBy: CodingKeys.self)
100
        
101
        try container.encodeIfPresent(globalAnalyticsConfiguration, forKey: .globalAnalyticsConfiguration)
102
        try container.encode(checkAgainTimeInterval, forKey: .checkAgainTimeInterval)
103
        try container.encode(queueServerDeploymentDestinations, forKey: .queueServerDeploymentDestinations)
104
        try container.encode(queueServerTerminationPolicy, forKey: .queueServerTerminationPolicy)
105
        try container.encode(workerDeploymentDestinations, forKey: .workerDeploymentDestinations)
106
        try container.encodeIfPresent(defaultWorkerConfiguration, forKey: .defaultWorkerConfiguration)
107
        try container.encode(workerStartMode, forKey: .workerStartMode)
108
        try container.encode(useOnlyIPv4, forKey: .useOnlyIPv4)
109
        try container.encode(portRange, forKey: .portRange)
110
    }
111
    
112
    public func workerConfiguration(
113
        workerSpecificConfiguration: WorkerSpecificConfiguration,
114
        payloadSignature: PayloadSignature
115
    ) -> WorkerConfiguration {
116
        return WorkerConfiguration(
117
            globalAnalyticsConfiguration: globalAnalyticsConfiguration,
118
            numberOfSimulators: workerSpecificConfiguration.numberOfSimulators,
119
            payloadSignature: payloadSignature,
120
            maximumCacheSize: workerSpecificConfiguration.maximumCacheSize,
121
            maximumCacheTTL: workerSpecificConfiguration.maximumCacheTTL,
122
            portRange: portRange
123
        )
124
    }
125
}
126