bmc/event_queue_thread.cc

#include <deque>
#include <iterator>
#include <memory>
#include <utility>
#include <vector>

#include "event.h"
#include "event_queue.h"
#include "absl/base/thread_annotations.h"
#include "absl/log/log.h"
#include "absl/status/status.h"
#include "absl/status/statusor.h"
#include "absl/synchronization/mutex.h"
#include "absl/time/time.h"

namespace safepower_agent {

struct EventQueue::Impl {
  absl::Mutex mutex;
  std::deque<Event> queue ABSL_GUARDED_BY(mutex);
  bool closed ABSL_GUARDED_BY(mutex) = false;

  bool NotEmptyOrClosed() const ABSL_SHARED_LOCKS_REQUIRED(mutex) {
    return !queue.empty() || closed;
  }
};

EventQueue::EventQueue() : impl_(std::make_unique<Impl>()) {}
EventQueue::~EventQueue() = default;

void EventQueue::Push(Event event) {
  absl::MutexLock lock(impl_->mutex);
  if (impl_->closed) {
    LOG(DFATAL) << "Attempted to push to a closed EventQueue.";
    return;
  }
  impl_->queue.push_back(std::move(event));
}

void EventQueue::Close() {
  absl::MutexLock lock(impl_->mutex);
  impl_->closed = true;
}

absl::StatusOr<std::vector<Event>> EventQueue::PopAllWithTimeout(
    absl::Duration timeout) {
  absl::MutexLock lock(impl_->mutex);

  if (impl_->queue.empty() && !impl_->closed &&
      timeout > absl::ZeroDuration()) {
    impl_->mutex.AwaitWithTimeout(
        absl::Condition(impl_.get(), &Impl::NotEmptyOrClosed), timeout);
  }

  if (impl_->queue.empty() && impl_->closed) {
    return absl::AbortedError("Event queue closed");
  }

  std::vector<Event> events(std::make_move_iterator(impl_->queue.begin()),
                            std::make_move_iterator(impl_->queue.end()));
  impl_->queue.clear();

  return events;
}

}  // namespace safepower_agent