11//
22// Copyright The OpenTelemetry Authors
33// SPDX-License-Identifier: Apache-2.0
4- //
4+ //
55
66import Foundation
77import OpenTelemetryApi
88
99public class BatchLogRecordProcessor : LogRecordProcessor {
10-
11-
10+
1211 fileprivate var worker : BatchWorker
13-
12+
1413 public init ( logRecordExporter: LogRecordExporter , scheduleDelay: TimeInterval = 5 , exportTimeout: TimeInterval = 30 , maxQueueSize: Int = 2048 , maxExportBatchSize: Int = 512 , willExportCallback: ( ( inout [ ReadableLogRecord ] ) -> Void ) ? = nil ) {
1514 worker = BatchWorker ( logRecordExporter: logRecordExporter, scheduleDelay: scheduleDelay, exportTimeout: exportTimeout, maxQueueSize: maxQueueSize, maxExportBatchSize: maxExportBatchSize, willExportCallback: willExportCallback)
16-
15+
1716 worker. start ( )
1817 }
19-
18+
2019 public func onEmit( logRecord: ReadableLogRecord ) {
2120 worker. emit ( logRecord: logRecord)
2221 }
23-
22+
2423 public func forceFlush( explicitTimeout: TimeInterval ? ) -> ExportResult {
2524 forceFlush ( timeout: explicitTimeout)
2625 return . success
2726 }
28-
27+
2928 public func forceFlush( timeout: TimeInterval ? = nil ) {
3029 worker. forceFlush ( explicitTimeout: timeout)
3130 }
32-
33-
31+
32+
3433 public func shutdown( explicitTimeout: TimeInterval ? = nil ) -> ExportResult {
3534 worker. cancel ( )
3635 worker. shutdown ( explicitTimeout: explicitTimeout)
3736 return . success
3837 }
3938}
4039
41- private class BatchWorker : Thread {
40+ private class BatchWorker {
41+ var thread : Thread !
4242 let logRecordExporter : LogRecordExporter
4343 let scheduleDelay : TimeInterval
4444 let maxQueueSize : Int
@@ -49,14 +49,14 @@ private class BatchWorker : Thread {
4949 private let cond = NSCondition ( )
5050 var logRecordList = [ ReadableLogRecord] ( )
5151 var queue : OperationQueue
52-
52+
5353 init ( logRecordExporter: LogRecordExporter ,
5454 scheduleDelay: TimeInterval ,
5555 exportTimeout: TimeInterval ,
5656 maxQueueSize: Int ,
5757 maxExportBatchSize: Int ,
5858 willExportCallback: ( ( inout [ ReadableLogRecord ] ) -> Void ) ? ) {
59-
59+
6060 self . logRecordExporter = logRecordExporter
6161 self . scheduleDelay = scheduleDelay
6262 self . exportTimeout = exportTimeout
@@ -67,58 +67,69 @@ private class BatchWorker : Thread {
6767 queue = OperationQueue ( )
6868 queue. name = " BatchWorker Queue "
6969 queue. maxConcurrentOperationCount = 1
70+ self . thread = Thread ( block: { [ weak self] in
71+ self ? . main ( )
72+ } )
73+ }
74+
75+ func start( ) {
76+ self . thread. start ( )
77+ }
78+
79+ func cancel( ) {
80+ self . thread. cancel ( )
7081 }
71-
82+
7283 func emit( logRecord: ReadableLogRecord ) {
7384 cond. lock ( )
74- defer { cond. unlock ( ) }
85+ defer { cond. unlock ( ) }
7586 if logRecordList. count == maxQueueSize {
7687 // TODO: record a counter for dropped logs
7788 return
7889 }
79-
90+
8091 // TODO: record a gauge for referenced logs
8192 logRecordList. append ( logRecord)
8293 if logRecordList. count >= halfMaxQueueSize {
8394 cond. broadcast ( )
8495 }
8596 }
86-
87- override func main( ) {
97+
98+ func main( ) {
8899 repeat {
89100 autoreleasepool {
90101 var logRecordsCopy : [ ReadableLogRecord ]
91102 cond. lock ( )
92103 if logRecordList. count < maxExportBatchSize {
93104 repeat {
94105 cond. wait ( until: Date ( ) . addingTimeInterval ( scheduleDelay) )
95- } while logRecordList. isEmpty && !self . isCancelled
106+ } while logRecordList. isEmpty && !thread . isCancelled
96107 }
97108 logRecordsCopy = logRecordList
98109 logRecordList. removeAll ( )
99110 cond. unlock ( )
100111 self . exportBatch ( logRecordList: logRecordsCopy, explicitTimeout: exportTimeout)
101112 }
102- } while !self . isCancelled
113+ } while !thread . isCancelled
103114 }
104-
115+
105116 public func forceFlush( explicitTimeout: TimeInterval ? = nil ) {
106117 var logRecordsCopy : [ ReadableLogRecord ]
107118 cond. lock ( )
108119 logRecordsCopy = logRecordList
109120 logRecordList. removeAll ( )
110121 cond. unlock ( )
111-
122+
112123 exportBatch ( logRecordList: logRecordsCopy, explicitTimeout: explicitTimeout)
113124 }
114-
115-
125+
126+
116127 public func shutdown( explicitTimeout: TimeInterval ? ) {
117128 let timeout = min ( explicitTimeout ?? TimeInterval . greatestFiniteMagnitude, exportTimeout)
118129 forceFlush ( explicitTimeout: timeout)
119130 _ = logRecordExporter. shutdown ( explicitTimeout: timeout)
120131 }
121-
132+
122133 private func exportBatch( logRecordList: [ ReadableLogRecord ] , explicitTimeout: TimeInterval ? = nil ) {
123134 let exportOperation = BlockOperation { [ weak self] in
124135 self ? . exportAction ( logRecordList : logRecordList, explicitTimeout: explicitTimeout)
@@ -132,7 +143,7 @@ private class BatchWorker : Thread {
132143 queue. waitUntilAllOperationsAreFinished ( )
133144 timeoutTimer. cancel ( )
134145 }
135-
146+
136147 private func exportAction( logRecordList: [ ReadableLogRecord ] , explicitTimeout: TimeInterval ? = nil ) {
137148 stride ( from: 0 , to: logRecordList. endIndex, by: maxExportBatchSize) . forEach {
138149 var logRecordToExport = logRecordList [ $0 ..< min ( $0 + maxExportBatchSize, logRecordList. count) ] . map { $0}
0 commit comments