| 98 | } |
| 99 | |
| 100 | void AsyncWaitFreeProcessor::ProcessThread(const std::function<void ()>& on_thread_initialize, const std::function<void ()>& on_thread_clenup) |
| 101 | { |
| 102 | // Call the thread initialize handler |
| 103 | assert((on_thread_initialize) && "Thread initialize handler must be valid!"); |
| 104 | if (on_thread_initialize) |
| 105 | on_thread_initialize(); |
| 106 | |
| 107 | try |
| 108 | { |
| 109 | // Thread local logger record to process |
| 110 | thread_local Record record; |
| 111 | thread_local uint64_t previous = CppCommon::Timestamp::utc(); |
| 112 | |
| 113 | while (_started) |
| 114 | { |
| 115 | // Try to dequeue the next logging record |
| 116 | bool empty = !_queue.Dequeue(record); |
| 117 | |
| 118 | // Current timestamp |
| 119 | uint64_t current; |
| 120 | |
| 121 | if (!empty) |
| 122 | { |
| 123 | // Handle stop operation record |
| 124 | if (record.timestamp == 0) |
| 125 | return; |
| 126 | |
| 127 | // Handle flush operation record |
| 128 | if (record.timestamp == 1) |
| 129 | { |
| 130 | // Flush the logging processor |
| 131 | Processor::Flush(); |
| 132 | continue; |
| 133 | } |
| 134 | |
| 135 | // Process logging record |
| 136 | Processor::ProcessRecord(record); |
| 137 | |
| 138 | // Update the current timestamp |
| 139 | current = record.timestamp; |
| 140 | } |
| 141 | else |
| 142 | { |
| 143 | // Update the current timestamp |
| 144 | current = CppCommon::Timestamp::utc(); |
| 145 | } |
| 146 | |
| 147 | // Handle auto-flush period |
| 148 | if (CppCommon::Timespan((int64_t)(current - previous)).seconds() > 1) |
| 149 | { |
| 150 | // Flush the logging processor |
| 151 | Processor::Flush(); |
| 152 | |
| 153 | // Update the previous timestamp |
| 154 | previous = current; |
| 155 | } |
| 156 | |
| 157 | // Sleep for a while if the queue was empty |