Emcee

1
import EventBus
2
import FileSystem
3
import Foundation
4
import EmceeLogging
5
import PathLib
6
import PluginSupport
7
import ProcessController
8
import ResourceLocationResolver
9
import SynchronousWaiter
10

11
public final class PluginManager: EventStream {
12
    private let encoder = JSONEncoder.pretty()
13
    private let eventDistributor: EventDistributor
14
    private let fileSystem: FileSystem
15
    private let logger: ContextualLogger
16
    private let pluginLocations: Set<AppleTestPluginLocation>
17
    private let pluginsConnectionTimeout: TimeInterval = 30.0
18
    private let processControllerProvider: ProcessControllerProvider
19
    private let resourceLocationResolver: ResourceLocationResolver
20
    private let sessionId = UUID()
21
    private let tearDownAllowance: TimeInterval = 60.0
22
    private var processControllers = [ProcessController]()
23
    
24
    public static let pluginBundleExtension = "emceeplugin"
25
    public static let pluginExecutableName = "Plugin"
26
    
27
    public init(
28
        fileSystem: FileSystem,
29
        logger: ContextualLogger,
30
        hostname: String,
31
        pluginLocations: Set<AppleTestPluginLocation>,
32
        processControllerProvider: ProcessControllerProvider,
33
        resourceLocationResolver: ResourceLocationResolver
34
    ) {
35
        self.fileSystem = fileSystem
36
        self.logger = logger
37
            .withMetadata(key: "pluginSessionId", value: sessionId.uuidString)
38
        self.eventDistributor = EventDistributor(hostname: hostname, logger: logger)
39
        self.pluginLocations = pluginLocations
40
        self.processControllerProvider = processControllerProvider
41
        self.resourceLocationResolver = resourceLocationResolver
42
    }
43
    
44
    private func pathsToPluginBundles() throws -> [AbsolutePath] {
45
        var paths = [AbsolutePath]()
46
        for location in pluginLocations {
47
            let resolvableLocation = resourceLocationResolver.resolvable(withRepresentable: location)
48
            
49
            let validatePathToPluginBundle: (AbsolutePath) throws -> () = { path in
50
                guard path.lastComponent.pathExtension == PluginManager.pluginBundleExtension else {
51
                    throw ValidationError.unexpectedExtension(location, actual: path.lastComponent.pathExtension, expected: PluginManager.pluginBundleExtension)
52
                }
53
                let executablePath = path.appending(PluginManager.pluginExecutableName)
54
                
55
                guard try self.fileSystem.properties(forFileAtPath: executablePath).isExecutable() else {
56
                    throw ValidationError.noExecutableFound(location, expectedLocation: executablePath)
57
                }
58
                
59
                paths.append(path)
60
            }
61
            
62
            switch try resolvableLocation.resolve() {
63
            case .directlyAccessibleFile(let path):
64
                try validatePathToPluginBundle(path)
65
            case .contentsOfArchive(let containerPath, let concretePluginName):
66
                if let concretePluginName = concretePluginName {
67
                    var path = containerPath.appending(concretePluginName)
68
                    if path.lastComponent.pathExtension != PluginManager.pluginBundleExtension {
69
                        path = path.appending(extension: PluginManager.pluginBundleExtension)
70
                    }
71
                    try validatePathToPluginBundle(path)
72
                } else {
73
                    var availablePlugins = [AbsolutePath]()
74
                    try fileSystem.contentEnumerator(forPath: containerPath, style: .shallow).each { path in
75
                        if path.extension == PluginManager.pluginBundleExtension {
76
                            availablePlugins.append(path)
77
                        }
78
                    }
79
                    
80
                    guard !availablePlugins.isEmpty else { throw ValidationError.noPluginsFound(location) }
81
                    for path in availablePlugins {
82
                        try validatePathToPluginBundle(path)
83
                    }
84
                }
85
            }
86
        }
87
        return paths
88
    }
89
    
90
    public func startPlugins() throws {
91
        try eventDistributor.start()
92
        let pluginSocket = try eventDistributor.webSocketAddress()
93
        
94
        let pluginBundles = try pathsToPluginBundles()
95
        
96
        for bundlePath in pluginBundles {
97
            logger.trace("Starting plugin at \(bundlePath)")
98
            let pluginExecutable = bundlePath.appending(PluginManager.pluginExecutableName)
99
            let pluginIdentifier = try pluginExecutable.pathString.avito_sha256Hash()
100
            eventDistributor.add(pluginIdentifier: pluginIdentifier)
101
            let controller = try processControllerProvider.createProcessController(
102
                subprocess: Subprocess(
103
                    arguments: [pluginExecutable],
104
                    environment: environmentForLaunchingPlugin(
105
                        pluginSocket: pluginSocket,
106
                        pluginIdentifier: pluginIdentifier
107
                    )
108
                )
109
            )
110
            try controller.start()
111
            processControllers.append(controller)
112
        }
113
        
114
        do {
115
            try eventDistributor.waitForPluginsToConnect(timeout: pluginsConnectionTimeout)
116
        } catch {
117
            logger.error("Failed to start plugins, will not tear down")
118
            tearDown()
119
            throw error
120
        }
121
    }
122
    
123
    private func environmentForLaunchingPlugin(pluginSocket: String, pluginIdentifier: String) -> Environment {
124
        [
125
            PluginSupport.pluginSocketEnv: pluginSocket,
126
            PluginSupport.pluginIdentifierEnv: pluginIdentifier
127
        ]
128
    }
129
    
130
    private func killPlugins() {
131
        logger.trace("Killing plugins that are still alive")
132
        for controller in processControllers {
133
            controller.interruptAndForceKillIfNeeded()
134
        }
135
        processControllers.removeAll()
136
    }
137
    
138
    // MARK: - Event Stream
139
    
140
    public func process(event: BusEvent) {
141
        send(busEvent: event)
142
        switch event {
143
        case .appleRunnerEvent(let event):
144
            runnerEvent(event)
145
        case .tearDown:
146
            tearDown()
147
        }
148
    }
149
    
150
    private func runnerEvent(_ event: AppleRunnerEvent) {}
151
    
152
    private func tearDown() {
153
        do {
154
            try SynchronousWaiter().waitWhile(timeout: tearDownAllowance, description: "Tear down plugins") {
155
                processControllers.map { $0.isProcessRunning }.contains(true)
156
            }
157
            logger.trace("All plugins torn down successfully")
158
        } catch {
159
            killPlugins()
160
        }
161
        eventDistributor.stop()
162
    }
163
    
164
    // MARK: - Re-distributing events to the plugins
165
    
166
    private func send(busEvent: BusEvent) {
167
        do {
168
            let data = try encoder.encode(busEvent)
169
            sendData(data)
170
        } catch {
171
            logger.error("Failed to get data for \(busEvent) event: \(error)")
172
        }
173
    }
174
    
175
    private func sendData(_ data: Data) {
176
        eventDistributor.send(data: data)
177
    }
178
}
179