task_queue_producer.h

Go to the documentation of this file.

#ifndef SENSESP_SYSTEM_TASK_QUEUE_PRODUCER_H_
#define SENSESP_SYSTEM_TASK_QUEUE_PRODUCER_H_
 
#include "ReactESP.h"
#include "observablevalue.h"
 
namespace sensesp {
 
template <class T>
class TaskQueueProducer : public ObservableValue<T> {
 public:
  TaskQueueProducer(const T& value, reactesp::ReactESP* consumer_app = ReactESP::app, int queue_size = 1,
                    unsigned int poll_rate = 990)
      : ObservableValue<T>(value), queue_size_{queue_size} {
    queue_ = xQueueCreate(queue_size, sizeof(T));
    if (queue_ == NULL) {
      debugE("Failed to create queue");
    }
 
    // Create a repeat reaction that will poll the queue and emit the values
    consumer_app->onRepeatMicros(poll_rate, [this]() {
      T value;
      while (xQueueReceive(queue_, &value, 0) == pdTRUE) {
        this->emit(value);
      }
    });
  }
 
  TaskQueueProducer(const T& value, int queue_size = 1, unsigned int poll_rate = 990)
      : TaskQueueProducer(value, ReactESP::app, queue_size, poll_rate) {}
 
  bool set(const T& value) {
    int retval;
    if (queue_size_ == 1) {
      retval = xQueueOverwrite(queue_, &value);
    } else {
      retval = xQueueSend(queue_, &value, 0);
    }
    if (retval != pdTRUE) {
      return false;
    }
    return true;
  }
 
 private:
  int queue_size_;
  QueueHandle_t queue_;
};
 
}  // namespace sensesp
 
#endif  // SENSESP_SYSTEM_TASK_QUEUE_PRODUCER_H_