| 176 | } // namespace detail |
| 177 | |
| 178 | inline UDPSender::UDPSender(const std::string& host, |
| 179 | const uint16_t port, |
| 180 | const uint64_t batchsize, |
| 181 | const uint64_t sendInterval) noexcept |
| 182 | : m_host(host), m_port(port), m_batchsize(batchsize), m_sendInterval(sendInterval) { |
| 183 | // Initialize the socket |
| 184 | if (!initialize()) { |
| 185 | return; |
| 186 | } |
| 187 | |
| 188 | // If batching is on, use a dedicated thread to send after the wait time is reached |
| 189 | if (m_batchsize != 0 && m_sendInterval > 0) { |
| 190 | // Define the batching thread |
| 191 | m_batchingThread = std::thread([this] { |
| 192 | // TODO: this will drop unsent stats, should we send all the unsent stats before we exit? |
| 193 | while (!m_mustExit.load(std::memory_order_acquire)) { |
| 194 | std::deque<std::string> stagedMessageQueue; |
| 195 | |
| 196 | std::unique_lock<std::mutex> batchingLock(m_batchingMutex); |
| 197 | m_batchingMessageQueue.swap(stagedMessageQueue); |
| 198 | batchingLock.unlock(); |
| 199 | |
| 200 | // Flush the queue |
| 201 | while (!stagedMessageQueue.empty()) { |
| 202 | sendToDaemon(stagedMessageQueue.front()); |
| 203 | stagedMessageQueue.pop_front(); |
| 204 | } |
| 205 | |
| 206 | // Wait before sending the next batch |
| 207 | std::this_thread::sleep_for(std::chrono::milliseconds(m_sendInterval)); |
| 208 | } |
| 209 | }); |
| 210 | } |
| 211 | } |
| 212 | |
| 213 | inline UDPSender::~UDPSender() { |
| 214 | if (!initialized()) { |