flush on home button

Author: abelonogov-ld

Sources/Common/Task+Extensions.swift

@@ -12,6 +12,6 @@ extension Task where Success == Never, Failure == Never {
     ///       try await Task.sleep(seconds: 3.0)
     ///
     public static func sleep(seconds: TimeInterval) async throws {
-        try await Task.sleep(nanoseconds: UInt64(seconds * Double(NSEC_PER_SEC)))
+        try await Task.sleep(nanoseconds: seconds.nanoseconds)
     }
 }

Sources/Common/Timestamp.swift

@@ -10,4 +10,8 @@ extension TimeInterval {
     public var milliseconds: Int64 {
         Int64(self * 1000.0)
     }
+    
+    public var nanoseconds: UInt64 {
+        UInt64(self * Double(NSEC_PER_SEC))
+    }
 }

Sources/LaunchDarklyObservability/Transport/BatchWorker.swift

@@ -8,7 +8,7 @@ public final actor BatchWorker {
     enum Constants {
         static let maxConcurrentCost: Int = 30000
         static let maxConcurrentItems: Int = 100
-        static let maxConcurrentExporters: Int = 2
+        static let maxConcurrentExporters: Int = 3
         static let baseBackoffSeconds: TimeInterval = 2
         static let maxBackoffSeconds: TimeInterval = 60
         static let backoffJitter: Double = 0.2
@@ -28,7 +28,8 @@ public final actor BatchWorker {
     private var costInFlight = 0
     private var exportersInFlight = Set<ObjectIdentifier>()
     private var exporterBackoff = [ObjectIdentifier: BackOffExporterInfo]()
-
+    private var flushableWorker: FlushableWorker?
+    
     public init(eventQueue: EventQueue,
                 log: OSLog) {
         self.eventQueue = eventQueue
@@ -40,27 +41,22 @@ public final actor BatchWorker {
         exporters[exporterId] = exporter
     }
 
-    public func start() {
-        guard task == nil else { return }
-        
-        task = Task.detached(priority: .background) { [weak self] in
-            while !Task.isCancelled {
-                guard let self else { return }
-                let scheduledCost = await sendQueueItems()
-                try? await Task.sleep(seconds: scheduledCost > 0 ? minInterval : interval)
-            }
+    public func start() async {
+        if flushableWorker == nil {
+            flushableWorker = FlushableWorker(interval: 1.5, work: sendQueueItems)
         }
+        await flushableWorker?.start()
     }
-    
-    func sendQueueItems() async -> Int {
+
+    func sendQueueItems(isFlushing: Bool) async {
         var scheduledCost = 0
 
         while true {
             let remainingExporterSlots = Constants.maxConcurrentExporters - exportersInFlight.count
-            guard remainingExporterSlots > 0 else { break }
+            guard remainingExporterSlots > 0 || isFlushing else { break }
 
             let budget = Constants.maxConcurrentCost - costInFlight
-            guard costInFlight == 0 || budget > 0 else { break }
+            guard costInFlight == 0 || (budget > 0 || isFlushing) else { break }
 
             let now = DispatchTime.now()
             var except = exportersInFlight
@@ -96,8 +92,6 @@ public final actor BatchWorker {
                 scheduledCost += cost
             }
         }
-        
-        return scheduledCost
     }
 
     private func tryReserve(exporterId: ObjectIdentifier, cost: Int) -> Bool {
@@ -128,8 +122,11 @@ public final actor BatchWorker {
         costInFlight -= cost
     }
 
-    public func stop() {
-        task?.cancel()
-        task = nil
+    public func stop() async {
+        await flushableWorker?.stop()
+    }
+    
+    public func flush() async {
+        await flushableWorker?.flush()
     }
 }

Sources/LaunchDarklyObservability/Transport/FlushableWorker.swift

@@ -0,0 +1,88 @@
+import Foundation
+#if !LD_COCOAPODS
+    import Common
+#endif
+
+actor FlushableWorker {
+    typealias Work = @Sendable (_ isFlushing: Bool) async -> Void
+    private enum Trigger {
+        case tick
+        case flush
+    }
+    
+    private var task: Task<Void, Never>? = nil
+    private let interval: TimeInterval
+    private var work: Work
+    private var continuation: AsyncStream<Trigger>.Continuation? = nil
+    private var pending: Trigger? = nil
+    
+    init(interval: TimeInterval, work: @escaping Work) {
+        self.interval = interval
+        self.work = work
+    }
+    
+    func start() {
+        guard task == nil else { return }
+        
+        let stream = AsyncStream<Trigger>(bufferingPolicy: .bufferingNewest(1)) { [weak self] cont in
+            Task {
+                await self?.setContinuation(cont)
+            }
+        }
+        
+        let task = Task { [weak self] in
+            guard let self else { return }
+            
+            let tickTask = Task {
+                while !Task.isCancelled {
+                    try? await Task.sleep(seconds: interval)
+                    doTrigger(.tick)
+                }
+            }
+            defer {
+                tickTask.cancel()
+            }
+
+            for await trigger in stream {
+                if Task.isCancelled {
+                    break
+                }
+                await work(trigger == .flush)
+                await clearPending()
+            }
+        }
+    }
+    
+    func stop() {
+        task?.cancel()
+        task = nil
+        continuation?.finish()
+        continuation = nil
+        pending = nil
+    }
+
+    func flush() async {
+        doTrigger(.flush)
+    }
+    
+    private func doTrigger(_ next: Trigger) {
+        guard pending != .flush else {
+            // flush is already next
+            return
+        }
+        
+        if next == .flush || pending == nil {
+            pending = next
+            continuation?.yield(next)
+        }
+    }
+    
+    private func clearPending() {
+        pending = nil
+    }
+    
+    private func setContinuation(_ continuation: AsyncStream<Trigger>.Continuation) {
+        self.continuation = continuation
+    }
+}
+

Sources/LaunchDarklyObservability/Transport/TransportService.swift

@@ -1,4 +1,5 @@
 import Foundation
+import Combine
 
 public protocol EventSource: AnyObject {
     func start()
@@ -16,13 +17,31 @@ final class TransportService: TransportServicing {
     public let eventQueue: EventQueue
     public let sessionManager: SessionManaging
     public private(set) var isRunnung: Bool = false
-    
     public var batchWorker: BatchWorker
-    
-    public init(eventQueue: EventQueue, batchWorker: BatchWorker, sessionManager: SessionManaging) {
+    private let appLifecycleManager: AppLifecycleManaging
+    private var cancellables = Set<AnyCancellable>()
+
+    public init(eventQueue: EventQueue,
+                batchWorker: BatchWorker,
+                sessionManager: SessionManaging,
+                appLifecycleManager: AppLifecycleManaging) {
         self.eventQueue = eventQueue
         self.batchWorker = batchWorker
         self.sessionManager = sessionManager
+        self.appLifecycleManager = appLifecycleManager
+        
+        
+        appLifecycleManager
+            .publisher()
+            .receive(on: DispatchQueue.global())
+            .sink { [weak self] event in
+                if event == .willResignActive {
+                    Task { [weak self] in
+                        await self?.batchWorker.flush()
+                    }
+                }
+            }
+            .store(in: &cancellables)
     }
 
     public func start() {