fix(logging): update flushlog to be blocking end to end (#3113)

* fix(logging): update flushlog to be blocking end to end

* chore(logging): fix typo

* chore(logging): address review comments

Author: phantumcode

AmplifyPlugins/Logging/Sources/AWSCloudWatchLoggingPlugin/AWSCloudWatchLoggingSessionController.swift

@@ -164,7 +164,22 @@ final class AWSCloudWatchLoggingSessionController {
     }
 
     func flushLogs() async throws {
-        try await session?.logger.flushLogs()
+        guard let logBatches = try await session?.logger.getLogBatches() else { return }
+                
+        for batch in logBatches {
+            try await consumeLogBatch(batch)
+        }
+    }
+    
+    private func consumeLogBatch(_ batch: LogBatch) async throws {
+        do {
+            try await consumer?.consume(batch: batch)
+        } catch {
+            Amplify.Logging.default.error("Error flushing logs with error \(error.localizedDescription)")
+            let payload = HubPayload(eventName: HubPayload.EventName.Logging.flushLogFailure, context: error.localizedDescription)
+            Amplify.Hub.dispatch(to: HubChannel.logging, payload: payload)
+            try batch.complete()
+        }
     }
 
     private func resetCurrentLogs() {

AmplifyPlugins/Logging/Sources/AWSCloudWatchLoggingPlugin/Persistence/LogActor.swift

@@ -50,11 +50,8 @@ actor LogActor {
         try rotation.currentLogFile.synchronize()
     }
 
-    func flushLogs() throws {
-        let logs = try rotation.getAllLogs()
-        for log in logs {
-            rotationSubject.send(log)
-        }
+    func getLogs() throws -> [URL] {
+        return try rotation.getAllLogs()
     }
 
     func deleteLogs() throws {

AmplifyPlugins/Logging/Sources/AWSCloudWatchLoggingPlugin/Producer/RotatingLogger.swift

@@ -43,9 +43,9 @@ final class RotatingLogger {
         try await actor.synchronize()
     }
 
-    func flushLogs() async throws {
-        try await setupSubscription()
-        try await actor.flushLogs()
+    func getLogBatches() async throws -> [RotatingLogBatch] {
+        let logs = try await actor.getLogs()
+        return logs.map({RotatingLogBatch(url: $0)})
     }
 
     func resetLogs() async throws {

AmplifyPlugins/Logging/Tests/AWSCloudWatchLoggingPluginTests/LogActorTests.swift

@@ -109,4 +109,22 @@ final class LogActorTests: XCTestCase {
         XCTAssertTrue(contents.isEmpty)
     }
 
+    
+    /// Given: a LogActor with 1 existing log
+    /// When: get all logs is called after writing and rotating to a new log
+    /// Then: 2 log files are returned
+    func testLogActorReturnsLogList() async throws {
+        var logs = try await systemUnderTest.getLogs()
+        XCTAssertEqual(logs.count, 1)
+        let size = try LogEntry.minimumSizeForLogEntry(level: .error)
+        let numberOfEntries = (fileSizeLimitInBytes/size) + 1
+        let entries = (0..<numberOfEntries).map { LogEntry(category: "", namespace: nil, level: .error, message: "\($0)", created: .init(timeIntervalSince1970: Double($0))) }
+        for entry in entries {
+            try await systemUnderTest.record(entry)
+        }
+        try await systemUnderTest.synchronize()
+        
+        logs = try await systemUnderTest.getLogs()
+        XCTAssertEqual(logs.count, 2)
+    }
 }

AmplifyPlugins/Logging/Tests/AWSCloudWatchLoggingPluginTests/RotatingLoggerTests.swift

@@ -123,6 +123,22 @@ final class RotatingLoggerTests: XCTestCase {
         try assertSingleEntryWith(level: level, message: message)
     }
 
+    /// Given: a rotating logger with 1 existing log batch
+    /// When: get log batch is called after writing and rotating to new log file
+    /// Then: a list of log batches with a count of 2 is returned
+    func testRotatingLogReturnsLogBatches() async throws {
+        var logBatches = try await systemUnderTest.getLogBatches()
+        XCTAssertEqual(logBatches.count, 1)
+        let size = try LogEntry.minimumSizeForLogEntry(level: .error)
+        let recordsPerFile = (fileSizeLimitInBytes/size)
+        for _ in 0..<recordsPerFile {
+            try await systemUnderTest.record(level: .error, message: "Test message")
+        }
+        try await systemUnderTest.synchronize()
+        logBatches = try await systemUnderTest.getLogBatches()
+        XCTAssertEqual(logBatches.count, 2)
+    }
+    
     private func logWith(level: LogLevel, message: String) async throws {
         switch level {
         case .error: