|
| 1 | +import Foundation |
| 2 | +import GRPC |
| 3 | +import NIO |
| 4 | +import os |
| 5 | + |
| 6 | +@MainActor |
| 7 | +protocolFileSyncDaemon:ObservableObject{ |
| 8 | +varstate:DaemonState{get} |
| 9 | +func start()asyncthrows |
| 10 | +func stop()asyncthrows |
| 11 | +} |
| 12 | + |
| 13 | +@MainActor |
| 14 | +classMutagenDaemon:FileSyncDaemon{ |
| 15 | +privateletlogger=Logger(subsystem:Bundle.main.bundleIdentifier!, category:"mutagen") |
| 16 | + |
| 17 | +@Publishedvarstate:DaemonState=.stopped |
| 18 | + |
| 19 | +privatevarmutagenProcess:Process? |
| 20 | +privatevarmutagenPipe:Pipe? |
| 21 | +privateletmutagenPath:URL |
| 22 | +privateletmutagenDataDirectory:URL |
| 23 | +privateletmutagenDaemonSocket:URL |
| 24 | + |
| 25 | +privatevargroup:MultiThreadedEventLoopGroup? |
| 26 | +privatevarchannel:GRPCChannel? |
| 27 | +privatevarclient:Daemon_DaemonAsyncClient? |
| 28 | + |
| 29 | +init(){ |
| 30 | +#if arch(arm64) |
| 31 | + mutagenPath=Bundle.main.url(forResource:"mutagen-darwin-arm64", withExtension:nil)! |
| 32 | +#elseif arch(x86_64) |
| 33 | + mutagenPath=Bundle.main.url(forResource:"mutagen-darwin-amd64", withExtension:nil)! |
| 34 | +#else |
| 35 | +fatalError("unknown architecture") |
| 36 | +#endif |
| 37 | + mutagenDataDirectory=FileManager.default.urls( |
| 38 | + for:.applicationSupportDirectory, |
| 39 | + in:.userDomainMask |
| 40 | +).first!.appending(path:"Coder Desktop").appending(path:"Mutagen") |
| 41 | + mutagenDaemonSocket= mutagenDataDirectory.appending(path:"daemon").appending(path:"daemon.sock") |
| 42 | + // It shouldn't be fatal if the app was built without Mutagen embedded, |
| 43 | + // but file sync will be unavailable. |
| 44 | +if !FileManager.default.fileExists(atPath: mutagenPath.path){ |
| 45 | + logger.warning("Mutagen not embedded in app, file sync will be unavailable") |
| 46 | + state=.unavailable |
| 47 | +} |
| 48 | +} |
| 49 | + |
| 50 | +func start()asyncthrows{ |
| 51 | +if case.unavailable= state{return} |
| 52 | + |
| 53 | + // Stop an orphaned daemon, if there is one |
| 54 | +try?awaitconnect() |
| 55 | +try?awaitstop() |
| 56 | + |
| 57 | +(mutagenProcess, mutagenPipe)=createMutagenProcess() |
| 58 | +do{ |
| 59 | +try mutagenProcess?.run() |
| 60 | +}catch{ |
| 61 | + state=.failed("Failed to start file sync daemon:\(error)") |
| 62 | +throwMutagenDaemonError.daemonStartFailure(error) |
| 63 | +} |
| 64 | + |
| 65 | +tryawaitconnect() |
| 66 | + |
| 67 | + state=.running |
| 68 | +} |
| 69 | + |
| 70 | +privatefunc connect()asyncthrows{ |
| 71 | +guard client==nilelse{ |
| 72 | + // Already connected |
| 73 | +return |
| 74 | +} |
| 75 | + group=MultiThreadedEventLoopGroup(numberOfThreads:1) |
| 76 | +do{ |
| 77 | + channel=tryGRPCChannelPool.with( |
| 78 | + target:.unixDomainSocket(mutagenDaemonSocket.path), |
| 79 | + transportSecurity:.plaintext, |
| 80 | + eventLoopGroup: group! |
| 81 | +) |
| 82 | + client=Daemon_DaemonAsyncClient(channel: channel!) |
| 83 | + logger.info("Successfully connected to mutagen daemon via gRPC") |
| 84 | +}catch{ |
| 85 | + logger.error("Failed to connect to gRPC:\(error)") |
| 86 | +tryawaitcleanupGRPC() |
| 87 | +throwMutagenDaemonError.connectionFailure(error) |
| 88 | +} |
| 89 | +} |
| 90 | + |
| 91 | +privatefunc cleanupGRPC()asyncthrows{ |
| 92 | +try?await channel?.close().get() |
| 93 | +try?await group?.shutdownGracefully() |
| 94 | + |
| 95 | + client=nil |
| 96 | + channel=nil |
| 97 | + group=nil |
| 98 | +} |
| 99 | + |
| 100 | +func stop()asyncthrows{ |
| 101 | +if case.unavailable= state{return} |
| 102 | + state=.stopped |
| 103 | +guardFileManager.default.fileExists(atPath: mutagenDaemonSocket.path)else{ |
| 104 | +return |
| 105 | +} |
| 106 | + |
| 107 | + // "We don't check the response or error, because the daemon |
| 108 | + // may terminate before it has a chance to send the response." |
| 109 | + _=try?await client?.terminate( |
| 110 | +Daemon_TerminateRequest(), |
| 111 | + callOptions:.init(timeLimit:.timeout(.milliseconds(500))) |
| 112 | +) |
| 113 | + |
| 114 | + // Clean up gRPC connection |
| 115 | +try?awaitcleanupGRPC() |
| 116 | + |
| 117 | + // Ensure the process is terminated |
| 118 | + mutagenProcess?.terminate() |
| 119 | + logger.info("Daemon stopped and gRPC connection closed") |
| 120 | +} |
| 121 | + |
| 122 | +privatefunc createMutagenProcess()->(Process,Pipe){ |
| 123 | +letoutputPipe=Pipe() |
| 124 | + outputPipe.fileHandleForReading.readabilityHandler= logOutput |
| 125 | +letprocess=Process() |
| 126 | + process.executableURL= mutagenPath |
| 127 | + process.arguments=["daemon","run"] |
| 128 | + process.environment=[ |
| 129 | +"MUTAGEN_DATA_DIRECTORY": mutagenDataDirectory.path, |
| 130 | +] |
| 131 | + process.standardOutput= outputPipe |
| 132 | + process.standardError= outputPipe |
| 133 | + process.terminationHandler= terminationHandler |
| 134 | +return(process, outputPipe) |
| 135 | +} |
| 136 | + |
| 137 | +privatenonisolatedfunc terminationHandler(process _:Process){ |
| 138 | +Task{@MainActorin |
| 139 | +self.mutagenPipe?.fileHandleForReading.readabilityHandler=nil |
| 140 | + mutagenProcess=nil |
| 141 | + |
| 142 | +try?awaitcleanupGRPC() |
| 143 | + |
| 144 | +switchself.state{ |
| 145 | +case.stopped: |
| 146 | + logger.info("mutagen daemon stopped") |
| 147 | +return |
| 148 | +default: |
| 149 | + logger.error("mutagen daemon exited unexpectedly") |
| 150 | +self.state=.failed("File sync daemon terminated unexpectedly") |
| 151 | +} |
| 152 | +} |
| 153 | +} |
| 154 | + |
| 155 | +privatenonisolatedfunc logOutput(pipe:FileHandle){ |
| 156 | +iflet line=String(data: pipe.availableData, encoding:.utf8), line!=""{ |
| 157 | + logger.info("\(line)") |
| 158 | +} |
| 159 | +} |
| 160 | +} |
| 161 | + |
| 162 | +enumDaemonState{ |
| 163 | +case running |
| 164 | +case stopped |
| 165 | +case failed(String) |
| 166 | +case unavailable |
| 167 | +} |
| 168 | + |
| 169 | +enumMutagenDaemonError:Error{ |
| 170 | +case daemonStartFailure(Error) |
| 171 | +case connectionFailure(Error) |
| 172 | +} |