KBF-36 implement thread-safe queue

include/kbf/queue.h

@@ -0,0 +1,54 @@
+#ifndef KBF_QUEUE_H
+#define KBF_QUEUE_H
+
+#include <optional>
+
+#include <freertos/FreeRTOS.h>
+#include <freertos/queue.h>
+
+#include "kbf/exception.h"
+
+namespace kbf {
+    template <typename T>
+    class Queue {
+    public:
+        explicit Queue(const unsigned int maxSize) : maxSize(maxSize) {
+            static_assert(!std::is_base_of<std::string, T>::value, "std::string is not supported");
+            if (!(handle = xQueueCreate(maxSize, sizeof(T)))) throw exception::OutOfMemoryError();
+        }
+
+        bool push(T value) {
+            return xQueueSendToBack(handle, ((void *) &value), 0) == pdTRUE;
+        }
+
+        std::optional<T> pop() {
+            T value;
+            if (xQueueReceive(handle, &value, portMAX_DELAY) == pdTRUE) return value;
+            else return std::nullopt;
+        }
+
+        std::optional<T> pop(int timeoutMs) {
+            T value;
+            if (xQueueReceive(handle, &value, timeoutMs / portTICK_PERIOD_MS) == pdTRUE) return value;
+            else return std::nullopt;
+        }
+
+        int size() {
+            return uxQueueMessagesWaiting(handle);
+        }
+
+        bool empty() {
+            return uxQueueMessagesWaiting(handle) == 0;
+        }
+
+        void clear() {
+            xQueueReset(handle);
+        }
+
+        const unsigned int maxSize;
+    private:
+        QueueHandle_t handle;
+    };
+}
+
+#endif //KBF_QUEUE_H

test/test_queue.cpp

@@ -0,0 +1,161 @@
+#include <vector>
+#include <string>
+
+#include "kbf.h"
+#include "kbf/exception.h"
+#include "kbf/queue.h"
+#include "kbf/task.h"
+
+#include <unity.h>
+
+#define DELAY 500  // milliseconds to sleep between pushes
+
+using namespace kbf;
+using std::vector;
+using std::string;
+
+template<typename T>
+struct PusherArg {
+    vector<T> &values;
+    Queue<T>  &queue;
+};
+
+template<typename T>
+class Pusher : public task::Task {
+public:
+    Pusher(const std::string &name, void *arg, uint32_t stackSize, uint32_t priority) :
+            Task(name, arg, stackSize, priority) {}
+
+protected:
+    void run(void *rawArgs) override {
+        auto args = static_cast<PusherArg<T> *>(rawArgs);
+
+        for (const auto &item : args->values) {
+            kbf::sleep(DELAY);
+            TEST_ASSERT_TRUE(args->queue.push(item));
+        }
+    }
+};
+
+TEST_CASE("Queue<int>", "[kbf_queue]") {
+    Queue<int>  queue(2);
+    TEST_ASSERT_EQUAL(2, queue.maxSize);
+    TEST_ASSERT_EQUAL(0, queue.size());
+    TEST_ASSERT_TRUE(queue.empty());
+
+    vector<int> values = {10, 20};
+    PusherArg<int> arg = {values, queue};
+    task::start<Pusher<int>>("test_queue", (void *) &arg);
+
+    kbf::sleep(DELAY * values.size() + 100);
+    TEST_ASSERT_EQUAL(values.size(), queue.size());
+
+    for (const auto &value : values) {
+        TEST_ASSERT_FALSE(queue.empty());
+        TEST_ASSERT_EQUAL(value, queue.pop().value());
+    }
+
+    TEST_ASSERT_TRUE(queue.pop(0) == std::nullopt);
+    TEST_ASSERT_TRUE(queue.empty());
+}
+
+//TEST_CASE("Queue<string>", "[kbf_queue]") {
+//    Queue<string>  queue(3);
+//    TEST_ASSERT_EQUAL(0, queue.size());
+//    TEST_ASSERT_TRUE(queue.empty());
+//
+//    vector<string> values = {"first", "second", "third"};
+//    PusherArg<string> arg = {values, queue};
+//    task::start<Pusher<string>>("test_queue", (void *) &arg);
+//
+//    kbf::sleep(DELAY * values.size() + 100);
+//    TEST_ASSERT_EQUAL(values.size(), queue.size());
+//
+//    for (const auto &value : values) {
+//        TEST_ASSERT_FALSE(queue.empty());
+//        TEST_ASSERT_EQUAL_STRING(value.c_str(), queue.pop().value().c_str());
+//    }
+//
+//    TEST_ASSERT_TRUE(queue.empty());
+//}
+
+TEST_CASE("Queue<struct>", "[kbf_queue]") {
+    struct Foo {
+        int a;
+        int b;
+    };
+
+    Queue<Foo>  queue(2);
+    TEST_ASSERT_EQUAL(0, queue.size());
+    TEST_ASSERT_TRUE(queue.empty());
+
+    vector<Foo> values = {{10, 20}, {30, 40}};
+    PusherArg<Foo> arg = {values, queue};
+    task::start<Pusher<Foo>>("test_queue", &arg);
+
+    kbf::sleep(DELAY * values.size() + 100);
+    TEST_ASSERT_EQUAL(values.size(), queue.size());
+
+    for (const auto &value : values) {
+        TEST_ASSERT_FALSE(queue.empty());
+        const std::optional<Foo> &current = queue.pop();
+        TEST_ASSERT_FALSE(current == std::nullopt);
+        TEST_ASSERT_EQUAL(value.a, current.value().a);
+        TEST_ASSERT_EQUAL(value.b, current.value().b);
+    }
+
+    TEST_ASSERT_TRUE(queue.empty());
+}
+
+TEST_CASE("Queue timeout", "[kbf_queue]") {
+    Queue<int>  queue(2);
+    TEST_ASSERT_EQUAL(0, queue.size());
+    TEST_ASSERT_TRUE(queue.empty());
+
+    vector<int> values = {10, 20};
+    PusherArg<int> arg = {values, queue};
+    task::start<Pusher<int>>("test_queue", (void *) &arg);
+
+    TEST_ASSERT_TRUE(queue.empty());
+    TEST_ASSERT_EQUAL(values[0], queue.pop(DELAY).value());
+
+    TEST_ASSERT_TRUE(queue.empty());
+    TEST_ASSERT_TRUE(queue.pop(DELAY / 2) == std::nullopt);
+
+    kbf::sleep(DELAY);
+    TEST_ASSERT_FALSE(queue.empty());
+    TEST_ASSERT_EQUAL(values[1], queue.pop().value());
+    TEST_ASSERT_TRUE(queue.empty());
+
+    TEST_ASSERT_TRUE(queue.pop(DELAY / 2) == std::nullopt);
+}
+
+TEST_CASE("Queue clear", "[kbf_queue]") {
+    Queue<int> queue(2);
+    TEST_ASSERT_EQUAL(0, queue.size());
+
+    TEST_ASSERT_TRUE(queue.push(10));
+    TEST_ASSERT_EQUAL(1, queue.size());
+
+    TEST_ASSERT_TRUE(queue.push(20));
+    TEST_ASSERT_EQUAL(2, queue.size());
+
+    queue.clear();
+    TEST_ASSERT_EQUAL(0, queue.size());
+}
+
+TEST_CASE("Queue overflow", "[kbf_queue]") {
+    Queue<int> queue(1);
+    TEST_ASSERT_EQUAL(1, queue.maxSize);
+    TEST_ASSERT_TRUE(queue.push(10));
+    TEST_ASSERT_FALSE(queue.push(20));
+}
+
+TEST_CASE("Queue out of memory", "[kbf_queue]") {
+    try {
+        [[maybe_unused]] Queue<int> queue(2 << 16);
+        TEST_FAIL();
+    } catch (exception::OutOfMemoryError &) {
+        return;
+    }
+}