log_queue.h

Go to the documentation of this file.

#ifndef SNIFF_LOG_QUEUE
#define SNIFF_LOG_QUEUE
 
#include "proto/service.pb.h"
#include <condition_variable>
#include <mutex>
#include <vector>
 
namespace yarilo {
 
class LogQueue {
public:
  LogQueue(uint64_t queueMaxSize) : queueMaxSize(queueMaxSize) {
    queue.reserve(queueMaxSize);
  }
 
  bool insert(proto::LogEntry *item) {
    {
      std::unique_lock<std::mutex> lock(queueMutex);
      if (!cvInsert.wait_for(lock, std::chrono::seconds(2),
                             [&queue = queue, &queueMaxSize = queueMaxSize] {
                               return queue.size() < queueMaxSize;
                             }))
        return false;
      queue.emplace_back(item);
    }
 
    cvFetch.notify_all();
    return true;
  }
 
  bool fetch_all(std::vector<proto::LogEntry *> &refFetchedItems) {
    {
      std::unique_lock<std::mutex> lock(queueMutex);
      if (!cvFetch.wait_for(lock, std::chrono::seconds(2),
                            [&queue = queue] { return !queue.empty(); }))
        return false;
      refFetchedItems = std::move(queue);
      queue.clear();
    }
 
    cvInsert.notify_all();
    return true;
  }
 
private:
  std::vector<proto::LogEntry *> queue;
  uint64_t queueMaxSize;
  std::mutex queueMutex;
  std::condition_variable cvFetch;
  std::condition_variable cvInsert;
};
 
} // namespace yarilo
 
#endif // SNIFF_LOG_QUEUE