feat(agent): implement send messages

Fixes ATL-2708

Castor/Tests/PeerDIDCreationTests.swift

@@ -62,12 +62,11 @@ final class PeerDIDCreationTests: XCTestCase {
         let mypeerDID = DID(
             schema: "did",
             method: "peer",
-            methodId: "2.Ez6LScuRnbZAJaVthcL5RELq75EBK2sBmBxsSs98LKNeriHQJ.Vz6MkpeaW7DeptJW7pi2qNXTdCXeQV4EYpZnAouH1LrfHj6uf.SeyJ0IjoiZG0iLCJzIjoiaHR0cHM6Ly9rOHMtZGV2LmF0YWxhcHJpc20uaW8vcHJpc20tYWdlbnQvZGlkY29tbSIsInIiOltdLCJhIjpbImRpZGNvbW0vdjIiXX0"
+            methodId: "2.Ez6LSfuJvTtcmcFrjNYYSAuD32tMZWQUD2HYDfXrJqy3ui6MQ.Vz6MkoPyvuxAecSezqmL4ERE7eW2XPbiUEHRH9aqay6LA8Eqr.SeyJ0IjoiZG0iLCJzIjoiaHR0cDovL2hvc3QuZG9ja2VyLmludGVybmFsOjgwL2RpZGNvbW0vIiwiciI6W10sImEiOlsiZGlkY29tbS92MiJdfQ"
         )
 
         let apollo = ApolloImpl()
         let castor = CastorImpl(apollo: apollo)
         let document = try await castor.resolveDID(did: mypeerDID)
-        print()
     }
 }

Core/Sources/Helpers/First+AsyncAwait.swift

@@ -0,0 +1,14 @@
+import Combine
+
+extension Publishers {
+    struct MissingOutputError: Error {}
+}
+
+public extension Publishers.First where Failure == Error {
+    func await() async throws -> Output {
+        for try await output in values {
+            return output
+        }
+        throw Publishers.MissingOutputError()
+    }
+}

PrismAgent/Sources/ConnectionsManager/ConnectionsManager.swift

@@ -137,6 +137,17 @@ class ConnectionsManagerImpl: ConnectionsManager {
 }
 
 extension ConnectionsManagerImpl: DIDCommConnection {
+    func sendMessage(_ message: Message) async throws -> Message? {
+        let mercury = self.mercury
+        return try await pluto
+            .storeMessage(message: message)
+            .flatMap {
+                Future { try await mercury.sendMessageParseMessage(msg: message) }
+            }
+            .first()
+            .await()
+    }
+
     func awaitMessages() async throws -> [Message] {
         let stream: AnyPublisher<[Message], Error> = try awaitMessages()
 

PrismAgent/Sources/ConnectionsManager/DIDCommProtocol.swift

@@ -4,6 +4,7 @@ import Foundation
 protocol DIDCommConnection {
     func awaitMessages() async throws -> [Message]
     func awaitMessageResponse(id: String) async throws -> Message?
+    func sendMessage(_ message: Message) async throws -> Message?
 }
 
 protocol ConnectionsManager {

PrismAgent/Sources/PrismAgent.swift

@@ -40,7 +40,7 @@ public class PrismAgent {
     private var connectionManager: ConnectionsManagerImpl
     private var cancellables = [AnyCancellable]()
     // Not a "stream"
-    private var messagesStreamTask: Task?
+    private var messagesStreamTask: Task<(), Error>?
 
     public let seed: Seed
 
@@ -318,6 +318,10 @@ public class PrismAgent {
         }
     }
 
+    public func sendMessage(_ message: Message) async throws -> Message? {
+        try await connectionManager.sendMessage(message)
+    }
+
     public func startFetchingMessages() {
         // TODO: This needs to be better thought for sure it cannot be left like this
         let manager = connectionManager

PrismAgent/Tests/Helper/DIDCommConnection+Testing.swift

@@ -4,6 +4,11 @@ import Domain
 class ConnectionStub: DIDCommConnection, ConnectionsManager {
     var awaitMessagesResponse: [Message]!
     var awaitMessageResponse: Message?
+    var sendMessageResponse: Message?
+
+    func sendMessage(_ message: Message) async throws -> Message? {
+        sendMessageResponse
+    }
 
     func awaitMessages() async throws -> [Message] {
         awaitMessagesResponse

Sample/AtalaPrismWalletDemo/AtalaPrismWalletDemo/Modules/SetupPrismAgent/SetupPrismAgentViewModel.swift

@@ -35,7 +35,7 @@ final class SetupPrismAgentViewModelImpl: ObservableObject, SetupPrismAgentViewM
     func updateKeyList() async throws {
         do {
             _ = try await agent.createNewPeerDID(updateMediator: true)
-            try await agent.awaitMessages()
+//            try await agent.awaitMessages()
         } catch {
             await MainActor.run {
                 self.error = error.localizedDescription