public:

explicit MessageQueue(uv_async_t *async)

: async_(async), data_(), shutdown_(false) {

uv_mutex_init(&lock_);

uv_cond_init(&cond_);

}

virtual ~MessageQueue() {

uv_cond_destroy(&cond_);

uv_mutex_destroy(&lock_);

}

inline uv_async_t *async() { return async_; }

void Push(Message *m) {

assert(m);

_Push(m);

}

void Notify() {

_Push(nullptr);

}

// Processes all messages on the queue.

// Returns true if at least one message was processed.

// If `match` is non null, it will block if the queue is empty and

// continue processing messages until `match->IsProcessed` is true.

template<typename Func>

bool DoProcess(Message *match, Func func) {

bool sawOne = false, loop = true;

Message *m;

while (loop) {

// Grab one message at a time, so that we don't end up processing

// messages out of order in case `func(m)` below ends up creating

// a recursive processing loop.

uv_mutex_lock(&lock_);

if (data_.empty()) {

m = nullptr;

if (match) {

// We're blocking for a particular message, and there's nothing here.

// Block to wait for some data.

uv_cond_wait(&cond_, &lock_);

} else {

loop = false;

}

} else {

sawOne = true;

m = data_.front();

data_.pop_front();

}

uv_mutex_unlock(&lock_);

if (m) { func(m); }

// Check whether either we processed the matching message,

// or else a recursive processing loop handled it for us.

if (match && match->IsProcessed()) { loop = false; }

}

return sawOne;

}

// Shutdown the queue: no more messages will be pushed

// after this method is called.

void Shutdown() {

uv_mutex_lock(&lock_);

shutdown_ = true;

uv_mutex_unlock(&lock_);

}

private:

void _Push(Message *m) {

bool was_shutdown = false;

uv_mutex_lock(&lock_);

if (!shutdown_) {

if (m) { data_.push_back(m); }

uv_cond_broadcast(&cond_);

// on a shutdown message, async_ could be torn down as soon

// as the other thread wakes up, so do the send inside the lock.

if (async_) { uv_async_send(async_); }

} else {

was_shutdown = true;

}

uv_mutex_unlock(&lock_);

if (was_shutdown) {

// Shouldn't happen (but assert after releasing the lock)

NPE_ERROR("Push after shutdown :(");

assert(false);

}

assert(!was_shutdown);

}

uv_async_t *async_;

uv_mutex_t lock_;

uv_cond_t cond_;

std::list<Message *> data_;

bool shutdown_;