| 201 | }; |
| 202 | |
| 203 | void TFileTransport::enqueueEvent(const uint8_t* buf, uint32_t eventLen) { |
| 204 | // can't enqueue more events if file is going to close |
| 205 | if (closing_) { |
| 206 | return; |
| 207 | } |
| 208 | |
| 209 | // make sure that event size is valid |
| 210 | if ((maxEventSize_ > 0) && (eventLen > maxEventSize_)) { |
| 211 | T_ERROR("msg size is greater than max event size: %u > %u\n", eventLen, maxEventSize_); |
| 212 | return; |
| 213 | } |
| 214 | |
| 215 | if (eventLen == 0) { |
| 216 | T_ERROR("%s", "cannot enqueue an empty event"); |
| 217 | return; |
| 218 | } |
| 219 | |
| 220 | std::unique_ptr<eventInfo, uniqueDeleter<eventInfo> > toEnqueue(new eventInfo()); |
| 221 | toEnqueue->eventBuff_ = new uint8_t[(sizeof(uint8_t) * eventLen) + 4]; |
| 222 | |
| 223 | // first 4 bytes is the event length |
| 224 | memcpy(toEnqueue->eventBuff_, (void*)(&eventLen), 4); |
| 225 | // actual event contents |
| 226 | memcpy(toEnqueue->eventBuff_ + 4, buf, eventLen); |
| 227 | toEnqueue->eventSize_ = eventLen + 4; |
| 228 | |
| 229 | // lock mutex |
| 230 | Guard g(mutex_); |
| 231 | |
| 232 | // make sure that enqueue buffer is initialized and writer thread is running |
| 233 | if (!bufferAndThreadInitialized_) { |
| 234 | if (!initBufferAndWriteThread()) { |
| 235 | return; |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | // Can't enqueue while buffer is full |
| 240 | while (enqueueBuffer_->isFull()) { |
| 241 | notFull_.wait(); |
| 242 | } |
| 243 | |
| 244 | // We shouldn't be trying to enqueue new data while a forced flush is |
| 245 | // requested. (Otherwise the writer thread might not ever be able to finish |
| 246 | // the flush if more data keeps being enqueued.) |
| 247 | assert(!forceFlush_); |
| 248 | |
| 249 | // add to the buffer |
| 250 | eventInfo* pEvent = toEnqueue.release(); |
| 251 | if (!enqueueBuffer_->addEvent(pEvent)) { |
| 252 | delete pEvent; |
| 253 | return; |
| 254 | } |
| 255 | |
| 256 | // signal anybody who's waiting for the buffer to be non-empty |
| 257 | notEmpty_.notify(); |
| 258 | |
| 259 | // this really should be a loop where it makes sure it got flushed |
| 260 | // because condition variables can get triggered by the os for no reason |