Support async in Command, MessageHandler and other core protocols

Package.resolved

@@ -337,8 +337,8 @@
       "kind" : "remoteSourceControl",
       "location" : "https://github.com/fwcd/swift-utils.git",
       "state" : {
-        "revision" : "fdd82aa5ab2e8da6102f997cee6bedee15f9aeba",
-        "version" : "3.0.5"
+        "revision" : "62e118fb9c4b2a0c0290f48b92f46d4a8f9f185e",
+        "version" : "3.0.7"
       }
     },
     {

Package.swift

@@ -18,7 +18,7 @@ let package = Package(
         .package(url: "https://github.com/apple/swift-argument-parser.git", from: "1.4.0"),
         .package(url: "https://github.com/fwcd/swift-qrcode-generator.git", from: "1.0.0"),
         .package(url: "https://github.com/fwcd/swift-prolog.git", from: "0.1.0"),
-        .package(url: "https://github.com/fwcd/swift-utils.git", from: "3.0.5"),
+        .package(url: "https://github.com/fwcd/swift-utils.git", from: "3.0.7"),
         .package(url: "https://github.com/fwcd/swift-graphics.git", from: "3.0.1"),
         .package(url: "https://github.com/fwcd/swift-gif.git", from: "3.1.0"),
         .package(url: "https://github.com/fwcd/swift-mensa.git", from: "0.1.10"),

Sources/D2Commands/ArgCommand.swift

@@ -7,16 +7,16 @@ public protocol ArgCommand: StringCommand {
     /// Fetches the _pattern instantation_ of the required argument format.
     var argPattern: Args { get }
 
-    func invoke(with input: Args, output: any CommandOutput, context: CommandContext)
+    func invoke(with input: Args, output: any CommandOutput, context: CommandContext) async
 }
 
 extension ArgCommand {
     public var inputValueType: RichValueType { .text }
 
-    public func invoke(with input: String, output: any CommandOutput, context: CommandContext) {
+    public func invoke(with input: String, output: any CommandOutput, context: CommandContext) async {
         let words = input.split(separator: " ").map { String($0) }
         if let args = Args.parse(from: TokenIterator(words)) {
-            invoke(with: args, output: output, context: context)
+            await invoke(with: args, output: output, context: context)
         } else {
             output.append(errorText: "Syntax: `\(argPattern)`")
         }

Sources/D2Commands/Command.swift

@@ -16,7 +16,7 @@ public protocol Command: AnyObject {
     ///
     /// Command invocations are inherently effectful and often asynchronous. This means
     /// that the passed output may be invoked on any thread, zero or (arbitrary) more times.
-    func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext)
+    func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) async
 
     /// Notifies the command that a message sent via CommandOutput has been
     /// successfully transmitted.

Sources/D2Commands/Meta/RandomCommand.swift

@@ -12,7 +12,7 @@ public class RandomCommand: Command {
         self.permissionManager = permissionManager
     }
 
-    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) {
+    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) async {
         guard let author = context.author else {
             output.append(errorText: "No author is present!")
             return
@@ -22,6 +22,6 @@ public class RandomCommand: Command {
             output.append(errorText: "No (permitted) commands found!")
             return
         }
-        command.invoke(with: input, output: output, context: context)
+        await command.invoke(with: input, output: output, context: context)
     }
 }

Sources/D2Commands/Meta/ReRunCommand.swift

@@ -9,14 +9,14 @@ public class ReRunCommand: VoidCommand {
         shouldOverwriteMostRecentPipeRunner: false
     )
     private let permissionManager: PermissionManager
-    @Synchronized @Box private var mostRecentPipeRunner: (Runnable, PermissionLevel)?
+    @Synchronized @Box private var mostRecentPipeRunner: (any AsyncRunnable, PermissionLevel)?
 
-    public init(permissionManager: PermissionManager, mostRecentPipeRunner: Synchronized<Box<(Runnable, PermissionLevel)?>>) {
+    public init(permissionManager: PermissionManager, mostRecentPipeRunner: Synchronized<Box<(any AsyncRunnable, PermissionLevel)?>>) {
         self.permissionManager = permissionManager
         self._mostRecentPipeRunner = mostRecentPipeRunner
     }
 
-    public func invoke(output: any CommandOutput, context: CommandContext) {
+    public func invoke(output: any CommandOutput, context: CommandContext) async {
         guard let (pipeRunner, minPermissionLevel) = mostRecentPipeRunner else {
             output.append(errorText: "No commands have been executed yet!")
             return
@@ -30,6 +30,6 @@ public class ReRunCommand: VoidCommand {
             return
         }
 
-        pipeRunner.run()
+        await pipeRunner.run()
     }
 }

Sources/D2Commands/Output/PipeOutput.swift

@@ -18,17 +18,20 @@ public class PipeOutput: CommandOutput {
     }
 
     public func append(_ value: RichValue, to channel: OutputChannel) {
-        let nextOutput = next ?? PrintOutput()
+        // TODO: Make append(_:to:) async
+        Task {
+            let nextOutput = next ?? PrintOutput()
 
-        if case .error(_, _) = value {
-            log.debug("Propagating error through pipe")
-            nextOutput.append(value, to: channel)
-        } else {
-            log.debug("Piping to \(sink)")
-            msgParser.parse(args, clientName: context.sink?.name, guild: context.guild).listenOrLogError {
-                let nextInput = $0 + value
-                log.trace("Invoking sink")
-                self.sink.invoke(with: nextInput, output: nextOutput, context: self.context)
+            if case .error(_, _) = value {
+                log.debug("Propagating error through pipe")
+                nextOutput.append(value, to: channel)
+            } else {
+                log.debug("Piping to \(sink)")
+                if let argsValue = await msgParser.parse(args, clientName: context.sink?.name, guild: context.guild).getOrLogError() {
+                    let nextInput = argsValue + value
+                    log.trace("Invoking sink")
+                    await self.sink.invoke(with: nextInput, output: nextOutput, context: self.context)
+                }
             }
         }
     }

Sources/D2Commands/Scripting/CronManager.swift

@@ -40,7 +40,7 @@ public class CronManager {
         }
     }
 
-    private func run(schedule: CronTab.Schedule) {
+    private func run(schedule: CronTab.Schedule) async {
         let parsedCommand = schedule.command
             .trimmingCharacters(in: .whitespacesAndNewlines)
             .split(separator: " ", maxSplits: 2, omittingEmptySubsequences: false)
@@ -54,7 +54,7 @@ public class CronManager {
             return
         }
 
-        msgParser.parse(commandArgs).listenOrLogError { [self] input in
+        if let input = await msgParser.parse(commandArgs).getOrLogError() {
             let context = CommandContext(
                 sink: sink,
                 registry: registry,
@@ -65,7 +65,7 @@ public class CronManager {
                 eventLoopGroup: eventLoopGroup
             )
             let output = MessageIOOutput(context: context)
-            command.invoke(with: input, output: output, context: context)
+            await command.invoke(with: input, output: output, context: context)
         }
     }
 
@@ -84,7 +84,9 @@ public class CronManager {
             let schedule = new[name]!
             do {
                 try scheduler.addSchedule(name: name, cron: schedule.cron) {
-                    self.run(schedule: schedule)
+                    Task {
+                        await self.run(schedule: schedule)
+                    }
                 }
                 liveCronTab.schedules[name] = schedule
             } catch {

Sources/D2Commands/StringCommand.swift

@@ -3,13 +3,13 @@ import D2Permissions
 /// A command that only expects text-based input (as opposed to e.g. an input embed).
 /// Usually, these are commands that expect exactly one argument.
 public protocol StringCommand: Command {
-    func invoke(with input: String, output: any CommandOutput, context: CommandContext)
+    func invoke(with input: String, output: any CommandOutput, context: CommandContext) async
 }
 
 extension StringCommand {
     public var inputValueType: RichValueType { .text }
 
-    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) {
-        invoke(with: input.asText ?? input.asCode ?? "", output: output, context: context)
+    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) async {
+        await invoke(with: input.asText ?? input.asCode ?? "", output: output, context: context)
     }
 }

Sources/D2Commands/VoidCommand.swift

@@ -2,13 +2,13 @@ import D2Permissions
 
 /// A command that expects no input.
 public protocol VoidCommand: Command {
-    func invoke(output: any CommandOutput, context: CommandContext)
+    func invoke(output: any CommandOutput, context: CommandContext) async
 }
 
 extension VoidCommand {
     public var inputValueType: RichValueType { .none }
 
-    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) {
-        invoke(output: output, context: context)
+    public func invoke(with input: RichValue, output: any CommandOutput, context: CommandContext) async {
+        await invoke(output: output, context: context)
     }
 }

Sources/D2Handlers/D2Receiver.swift

@@ -61,7 +61,7 @@ public class D2Receiver: Receiver {
         permissionManager = PermissionManager()
         let inventoryManager = InventoryManager()
 
-        @Synchronized @Box var mostRecentPipeRunner: (Runnable, PermissionLevel)? = nil
+        @Synchronized @Box var mostRecentPipeRunner: (any AsyncRunnable, PermissionLevel)? = nil
         @AutoSerializing(filePath: "local/spamConfig.json") var spamConfiguration = SpamConfiguration()
         @AutoSerializing(filePath: "local/streamerRoleConfig.json") var streamerRoleConfiguration = StreamerRoleConfiguration()
         @AutoSerializing(filePath: "local/messagePreviewsConfig.json") var messagePreviewsConfiguration = MessagePreviewsConfiguration()
@@ -570,27 +570,29 @@ public class D2Receiver: Receiver {
     }
 
     public func on(createMessage message: Message, sink: any Sink) {
-        var m = message
-
-        for rewriter in messageRewriters {
-            if let rewrite = rewriter.rewrite(message: m, sink: sink) {
-                m = rewrite
+        // TODO: Make on(createMessage:) itself (and the other methods) async in
+        // the protocol and remove this explicit Task.
+        Task {
+            var m = message
+
+            for rewriter in messageRewriters {
+                if let rewrite = await rewriter.rewrite(message: m, sink: sink) {
+                    m = rewrite
+                }
             }
-        }
 
-        for i in messageHandlers.indices {
-            if messageHandlers[i].handleRaw(message: message, sink: sink) {
-                return
-            }
-            if messageHandlers[i].handle(message: m, sink: sink) {
-                return
+            for i in messageHandlers.indices {
+                if await messageHandlers[i].handleRaw(message: message, sink: sink) {
+                    return
+                }
+                if await messageHandlers[i].handle(message: m, sink: sink) {
+                    return
+                }
             }
-        }
 
-        // Only fire on unhandled messages
-        if m.author?.id != sink.me?.id {
-            MessageParser().parse(message: m, clientName: sink.name, guild: m.guild).listenOrLogError {
-                self.eventListenerBus.fire(event: .createMessage, with: $0, context: CommandContext(
+            // Only fire on unhandled messages
+            if m.author?.id != sink.me?.id, let value = await MessageParser().parse(message: m, clientName: sink.name, guild: m.guild).getOrLogError() {
+                eventListenerBus.fire(event: .createMessage, with: value, context: CommandContext(
                     sink: sink,
                     registry: self.registry,
                     message: m,
@@ -604,9 +606,13 @@ public class D2Receiver: Receiver {
     }
 
     public func on(createInteraction interaction: Interaction, sink: any Sink) {
-        for i in interactionHandlers.indices {
-            if interactionHandlers[i].handle(interaction: interaction, sink: sink) {
-                return
+        // TODO: Make on(createInteraction:) itself (and the other methods)
+        // async in the protocol and remove this explicit Task.
+        Task {
+            for i in interactionHandlers.indices {
+                if await interactionHandlers[i].handle(interaction: interaction, sink: sink) {
+                    return
+                }
             }
         }
     }

Sources/D2Handlers/Interaction/Handler/InteractionHandler.swift

@@ -2,7 +2,7 @@ import D2MessageIO
 
 /// Anything that handles interactions from Discord.
 public protocol InteractionHandler {
-    mutating func handle(interaction: Interaction, sink: any Sink) -> Bool
+    mutating func handle(interaction: Interaction, sink: any Sink) async -> Bool
 }
 
 extension InteractionHandler {

Sources/D2Handlers/Interaction/Handler/MIOCommandInteractionHandler.swift

@@ -16,7 +16,7 @@ public struct MIOCommandInteractionHandler: InteractionHandler {
         self.eventLoopGroup = eventLoopGroup
     }
 
-    public func handle(interaction: Interaction, sink: any Sink) -> Bool {
+    public func handle(interaction: Interaction, sink: any Sink) async -> Bool {
         guard
             interaction.type == .mioCommand,
             let data = interaction.data,
@@ -54,7 +54,7 @@ public struct MIOCommandInteractionHandler: InteractionHandler {
             return true
         }
 
-        command.invoke(with: input, output: output, context: context)
+        await command.invoke(with: input, output: output, context: context)
         return true
     }
 }

Sources/D2Handlers/Message/Handler/CommandHandler.swift

@@ -25,12 +25,12 @@ fileprivate class PipeComponent {
     }
 }
 
-fileprivate struct RunnablePipe: Runnable {
+fileprivate struct RunnablePipe: AsyncRunnable {
     let pipeSource: PipeComponent
     let input: RichValue
 
-    func run() {
-        pipeSource.command.invoke(with: input, output: pipeSource.output!, context: pipeSource.context)
+    func run() async {
+        await pipeSource.command.invoke(with: input, output: pipeSource.output!, context: pipeSource.context)
     }
 }
 
@@ -57,7 +57,7 @@ public class CommandHandler: MessageHandler {
     private let unconditionallyAllowedCommands: Set<String>
 
     @Synchronized private var currentlyRunningCommands = 0
-    @Synchronized @Box private var mostRecentPipeRunner: (Runnable, PermissionLevel)?
+    @Synchronized @Box private var mostRecentPipeRunner: (any AsyncRunnable, PermissionLevel)?
 
     private let commandQueue = DispatchQueue(label: "CommandHandler", attributes: [.concurrent])
 
@@ -68,7 +68,7 @@ public class CommandHandler: MessageHandler {
         permissionManager: PermissionManager,
         subscriptionManager: SubscriptionManager,
         eventLoopGroup: any EventLoopGroup,
-        mostRecentPipeRunner: Synchronized<Box<(Runnable, PermissionLevel)?>>,
+        mostRecentPipeRunner: Synchronized<Box<(any AsyncRunnable, PermissionLevel)?>>,
         maxPipeLengthForUsers: Int = 7,
         maxConcurrentlyRunningCommands: Int = 4,
         unconditionallyAllowedCommands: Set<String> = ["quit"],
@@ -89,7 +89,7 @@ public class CommandHandler: MessageHandler {
         self.pipeSeparator = pipeSeparator
     }
 
-    public func handle(message: Message, sink: any Sink) -> Bool {
+    public func handle(message: Message, sink: any Sink) async -> Bool {
         guard message.content.starts(with: commandPrefix),
             !message.dm || (message.author.map { permissionManager.user($0, hasPermission: .vip) } ?? false),
             let channelId = message.channelId else { return false }
@@ -152,20 +152,23 @@ public class CommandHandler: MessageHandler {
                 guard let pipeSource = pipe.first else { continue }
 
                 commandQueue.async {
-                    self.currentlyRunningCommands += 1
-                    log.debug("Currently running \(self.currentlyRunningCommands) commands")
-
-                    self.msgParser.parse(pipeSource.args, message: message, clientName: sink.name, guild: pipeSource.context.guild).listenOrLogError { input in
-                        // Execute the pipe
-                        let runner = RunnablePipe(pipeSource: pipeSource, input: input)
-                        runner.run()
-
-                        // Store the pipe for potential re-execution
-                        if pipe.allSatisfy({ $0.command.info.shouldOverwriteMostRecentPipeRunner }), let minPermissionLevel = pipe.map(\.command.info.requiredPermissionLevel).max() {
-                            self.mostRecentPipeRunner = (runner, minPermissionLevel)
+                    // TODO: Remove commandQueue and make CommandHandler an actor instead?
+                    Task {
+                        self.currentlyRunningCommands += 1
+                        log.debug("Currently running \(self.currentlyRunningCommands) commands")
+
+                        if let input = await self.msgParser.parse(pipeSource.args, message: message, clientName: sink.name, guild: pipeSource.context.guild).getOrLogError() {
+                            // Execute the pipe
+                            let runner = RunnablePipe(pipeSource: pipeSource, input: input)
+                            await runner.run()
+
+                            // Store the pipe for potential re-execution
+                            if pipe.allSatisfy({ $0.command.info.shouldOverwriteMostRecentPipeRunner }), let minPermissionLevel = pipe.map(\.command.info.requiredPermissionLevel).max() {
+                                self.mostRecentPipeRunner = (runner, minPermissionLevel)
+                            }
+
+                            self.currentlyRunningCommands -= 1
                         }
-
-                        self.currentlyRunningCommands -= 1
                     }
                 }
             }

Sources/D2Handlers/Message/Handler/MessageHandler.swift

@@ -8,15 +8,15 @@ public protocol MessageHandler {
     /// Processes the message and returns whether it was handled (successfully).
     /// Handlers can also return false if they only "observed" the message, but
     /// did not intend to "consume" it.
-    mutating func handle(message: Message, sink: any Sink) -> Bool
+    mutating func handle(message: Message, sink: any Sink) async -> Bool
 
     /// Processes the raw (non-rewritten) message and returns whether it was handled
     /// (sucessfully). This method will always be invoked prior to the actual handle
     /// method.
     ///
     /// Generally, you should avoid implementing this method unless you have a good
     /// reason to do so, since this may cause unintended message semantics.
-    mutating func handleRaw(message: Message, sink: any Sink) -> Bool
+    mutating func handleRaw(message: Message, sink: any Sink) async -> Bool
 }
 
 public extension MessageHandler {

Sources/D2Handlers/Message/Rewriter/MessageRewriter.swift

@@ -2,5 +2,5 @@ import D2MessageIO
 
 /// Represents anything that modifies an (incoming) message.
 public protocol MessageRewriter {
-    func rewrite(message: Message, sink: any Sink) -> Message?
+    func rewrite(message: Message, sink: any Sink) async -> Message?
 }