Pubsub core api feature (#174)

* fixed inline functions for modern C standart

* pubsub api, base version

* basic test for pubsub

* update applications.mk, add test file

* more test for pubsub

* remove unimplemented files, cleanup header file

* remove legacy tests, check unsubscribe not call cb

* implement deleting mutex, fail test

* release mutex before deleting

Co-authored-by: aanper <mail@s3f.ru>

applications/applications.mk

@@ -26,6 +26,7 @@ C_SOURCES	+= $(APP_DIR)/tests/furi_record_test.c
 C_SOURCES	+= $(APP_DIR)/tests/test_index.c
 C_SOURCES	+= $(APP_DIR)/tests/minunit_test.c
 C_SOURCES	+= $(APP_DIR)/tests/furi_valuemutex_test.c
+C_SOURCES	+= $(APP_DIR)/tests/furi_pubsub_test.c
 C_SOURCES	+= $(APP_DIR)/tests/furi_memmgr_test.c
 endif
 

applications/tests/furi_pubsub_test.c

@@ -0,0 +1,56 @@
+#include <stdio.h>
+#include <string.h>
+#include "flipper_v2.h"
+#include "log.h"
+
+#include "minunit.h"
+
+const uint32_t context_value = 0xdeadbeef;
+const uint32_t notify_value_0 = 0x12345678;
+const uint32_t notify_value_1 = 0x11223344;
+
+uint32_t pubsub_value = 0;
+uint32_t pubsub_context_value = 0;
+
+void test_pubsub_handler(void* arg, void* ctx) {
+    pubsub_value = *(uint32_t*)arg;
+    pubsub_context_value = *(uint32_t*)ctx;
+}
+
+void test_furi_pubsub() {
+    bool result;
+    PubSub test_pubsub;
+    PubSubItem* test_pubsub_item;
+
+    // init pubsub case
+    result = init_pubsub(&test_pubsub);
+    mu_assert(result, "init pubsub failed");
+
+    // subscribe pubsub case
+    test_pubsub_item = subscribe_pubsub(&test_pubsub, test_pubsub_handler, (void*)&context_value);
+    mu_assert_pointers_not_eq(test_pubsub_item, NULL);
+
+    /// notify pubsub case
+    result = notify_pubsub(&test_pubsub, (void*)&notify_value_0);
+    mu_assert(result, "notify pubsub failed");
+    mu_assert_int_eq(pubsub_value, notify_value_0);
+    mu_assert_int_eq(pubsub_context_value, context_value);
+
+    // unsubscribe pubsub case
+    result = unsubscribe_pubsub(test_pubsub_item);
+    mu_assert(result, "unsubscribe pubsub failed");
+
+    result = unsubscribe_pubsub(test_pubsub_item);
+    mu_assert(!result, "unsubscribe pubsub not failed");
+
+    /// notify unsubscribed pubsub case
+    result = notify_pubsub(&test_pubsub, (void*)&notify_value_1);
+    mu_assert(result, "notify pubsub failed");
+    mu_assert_int_not_eq(pubsub_value, notify_value_1);
+
+    // delete pubsub case
+    result = delete_pubsub(&test_pubsub);
+    mu_assert(result, "unsubscribe pubsub failed");
+
+    // TODO test case that the pubsub_delete will remove pubsub from heap
+}

applications/tests/furi_record_test.c

@@ -14,197 +14,3 @@ void test_furi_create_open() {
     void* record = furi_open("test/holding");
     mu_assert_pointers_eq(record, &test_data);
 }
-
-/*
-TEST: non-existent data
-1. Try to open non-existent record
-2. Check for NULL handler
-3. Try to write/read, get error
-
-TODO: implement this test
-*/
-bool test_furi_nonexistent_data() {
-    return true;
-}
-
-/*
-TEST: mute algorithm
-1. Create "parent" application:
-    1. Create pipe record
-    2. Open watch handler: no_mute=false, solo=false, subscribe to data.
-
-2. Open handler A: no_mute=false, solo=false, NULL subscriber. Subscribe to state.
-Try to write data to A and check subscriber.
-
-3. Open handler B: no_mute=true, solo=true, NULL subscriber.
-Check A state cb get FlipperRecordStateMute.
-Try to write data to A and check that subscriber get no data. (muted)
-Try to write data to B and check that subscriber get data.
-
-TODO: test 3 not pass beacuse state callback not implemented
-
-4. Open hadler C: no_mute=false, solo=true, NULL subscriber.
-Try to write data to A and check that subscriber get no data. (muted)
-Try to write data to B and check that subscriber get data. (not muted because open with no_mute)
-Try to write data to C and check that subscriber get data.
-
-5. Open handler D: no_mute=false, solo=false, NULL subscriber.
-Try to write data to A and check that subscriber get no data. (muted)
-Try to write data to B and check that subscriber get data. (not muted because open with no_mute)
-Try to write data to C and check that subscriber get data. (not muted because D open without solo)
-Try to write data to D and check that subscriber get data.
-
-6. Close C, close B.
-Check A state cb get FlipperRecordStateUnmute
-Try to write data to A and check that subscriber get data. (unmuted)
-Try to write data to D and check that subscriber get data.
-
-TODO: test 6 not pass beacuse cleanup is not implemented
-TODO: test 6 not pass because mute algorithm is unfinished.
-
-7. Exit "parent application"
-Check A state cb get FlipperRecordStateDeleted
-
-TODO: test 7 not pass beacuse cleanup is not implemented
-*/
-
-static uint8_t mute_last_value = 0;
-static FlipperRecordState mute_last_state = 255;
-
-void mute_record_cb(const void* value, size_t size, void* ctx) {
-    // hold value to static var
-    mute_last_value = *((uint8_t*)value);
-}
-
-void mute_record_state_cb(FlipperRecordState state, void* ctx) {
-    mute_last_state = state;
-}
-
-void furi_mute_parent_app(void* p) {
-    // 1. Create pipe record
-    if(!furi_create_deprecated("test/mute", NULL, 0)) {
-        printf("cannot create record\n");
-        furiac_exit(NULL);
-    }
-
-    // 2. Open watch handler: solo=false, no_mute=false, subscribe to data
-    FuriRecordSubscriber* watch_handler =
-        furi_open_deprecated("test/mute", false, false, mute_record_cb, NULL, NULL);
-    if(watch_handler == NULL) {
-        printf("cannot open watch handler\n");
-        furiac_exit(NULL);
-    }
-
-    while(1) {
-        // TODO we don't have thread sleep
-        delay(100000);
-    }
-}
-
-bool test_furi_mute_algorithm() {
-    // 1. Create "parent" application:
-    FuriApp* parent_app = furiac_start(furi_mute_parent_app, "parent app", NULL);
-
-    delay(2); // wait creating record
-
-    // 2. Open handler A: solo=false, no_mute=false, NULL subscriber. Subscribe to state.
-    FuriRecordSubscriber* handler_a =
-        furi_open_deprecated("test/mute", false, false, NULL, mute_record_state_cb, NULL);
-    if(handler_a == NULL) {
-        printf("cannot open handler A\n");
-        return false;
-    }
-
-    uint8_t test_counter = 1;
-
-    // Try to write data to A and check subscriber
-    if(!furi_write(handler_a, &test_counter, sizeof(uint8_t))) {
-        printf("write to A failed\n");
-        return false;
-    }
-
-    if(mute_last_value != test_counter) {
-        printf("value A mismatch: %d vs %d\n", mute_last_value, test_counter);
-        return false;
-    }
-
-    // 3. Open handler B: solo=true, no_mute=true, NULL subscriber.
-    FuriRecordSubscriber* handler_b =
-        furi_open_deprecated("test/mute", true, true, NULL, NULL, NULL);
-    if(handler_b == NULL) {
-        printf("cannot open handler B\n");
-        return false;
-    }
-
-    // Check A state cb get FlipperRecordStateMute.
-    if(mute_last_state != FlipperRecordStateMute) {
-        printf("A state is not FlipperRecordStateMute: %d\n", mute_last_state);
-        return false;
-    }
-
-    test_counter = 2;
-
-    // Try to write data to A and check that subscriber get no data. (muted)
-    if(furi_write(handler_a, &test_counter, sizeof(uint8_t))) {
-        printf("A not muted\n");
-        return false;
-    }
-
-    if(mute_last_value == test_counter) {
-        printf("value A must be muted\n");
-        return false;
-    }
-
-    test_counter = 3;
-
-    // Try to write data to B and check that subscriber get data.
-    if(!furi_write(handler_b, &test_counter, sizeof(uint8_t))) {
-        printf("write to B failed\n");
-        return false;
-    }
-
-    if(mute_last_value != test_counter) {
-        printf("value B mismatch: %d vs %d\n", mute_last_value, test_counter);
-        return false;
-    }
-
-    // 4. Open hadler C: solo=true, no_mute=false, NULL subscriber.
-    FuriRecordSubscriber* handler_c =
-        furi_open_deprecated("test/mute", true, false, NULL, NULL, NULL);
-    if(handler_c == NULL) {
-        printf("cannot open handler C\n");
-        return false;
-    }
-
-    // TODO: Try to write data to A and check that subscriber get no data. (muted)
-    // TODO: Try to write data to B and check that subscriber get data. (not muted because open with no_mute)
-    // TODO: Try to write data to C and check that subscriber get data.
-
-    // 5. Open handler D: solo=false, no_mute=false, NULL subscriber.
-    FuriRecordSubscriber* handler_d =
-        furi_open_deprecated("test/mute", false, false, NULL, NULL, NULL);
-    if(handler_d == NULL) {
-        printf("cannot open handler D\n");
-        return false;
-    }
-
-    // TODO: Try to write data to A and check that subscriber get no data. (muted)
-    // TODO: Try to write data to B and check that subscriber get data. (not muted because open with no_mute)
-    // TODO: Try to write data to C and check that subscriber get data. (not muted because D open without solo)
-    // TODO: Try to write data to D and check that subscriber get data.
-
-    // 6. Close C, close B.
-    // TODO: Check A state cb get FlipperRecordStateUnmute
-    // TODO: Try to write data to A and check that subscriber get data. (unmuted)
-    // TODO: Try to write data to D and check that subscriber get data.
-
-    // 7. Exit "parent application"
-    if(!furiac_kill(parent_app)) {
-        printf("kill parent_app fail\n");
-        return false;
-    }
-
-    // TODO: Check A state cb get FlipperRecordStateDeleted
-
-    return true;
-}

applications/tests/minunit_test.c

@@ -7,13 +7,12 @@
 bool test_furi_ac_create_kill();
 bool test_furi_ac_switch_exit();
 
-bool test_furi_nonexistent_data();
-bool test_furi_mute_algorithm();
-
 // v2 tests
 void test_furi_create_open();
 void test_furi_valuemutex();
 void test_furi_concurrent_access();
+void test_furi_pubsub();
+
 void test_furi_memmgr();
 
 static int foo = 0;
@@ -38,10 +37,6 @@ MU_TEST(mu_test_furi_ac_switch_exit) {
     mu_assert_int_eq(test_furi_ac_switch_exit(), true);
 }
 
-MU_TEST(mu_test_furi_nonexistent_data) {
-    mu_assert_int_eq(test_furi_nonexistent_data(), true);
-}
-
 // v2 tests
 MU_TEST(mu_test_furi_create_open) {
     test_furi_create_open();
@@ -55,6 +50,10 @@ MU_TEST(mu_test_furi_concurrent_access) {
     test_furi_concurrent_access();
 }
 
+MU_TEST(mu_test_furi_pubsub) {
+    test_furi_pubsub();
+}
+
 MU_TEST(mu_test_furi_memmgr) {
     // this test is not accurate, but gives a basic understanding
     // that memory management is working fine
@@ -68,12 +67,11 @@ MU_TEST_SUITE(test_suite) {
     MU_RUN_TEST(mu_test_furi_ac_create_kill);
     MU_RUN_TEST(mu_test_furi_ac_switch_exit);
 
-    MU_RUN_TEST(mu_test_furi_nonexistent_data);
-
     // v2 tests
     MU_RUN_TEST(mu_test_furi_create_open);
     MU_RUN_TEST(mu_test_furi_valuemutex);
     MU_RUN_TEST(mu_test_furi_concurrent_access);
+    MU_RUN_TEST(mu_test_furi_pubsub);
 
     MU_RUN_TEST(mu_test_furi_memmgr);
 }

core/api-basic/pubsub.c

@@ -0,0 +1,90 @@
+#include "pubsub.h"
+
+bool init_pubsub(PubSub* pubsub) {
+    // mutex without name,
+    // no attributes (unfortunatly robust mutex is not supported by FreeRTOS),
+    // with dynamic memory allocation
+    const osMutexAttr_t value_mutex_attr = {
+        .name = NULL, .attr_bits = 0, .cb_mem = NULL, .cb_size = 0U};
+
+    pubsub->mutex = osMutexNew(&value_mutex_attr);
+    if(pubsub->mutex == NULL) return false;
+
+    // construct list
+    list_pubsub_cb_init(pubsub->items);
+
+    return true;
+}
+
+bool delete_pubsub(PubSub* pubsub) {
+    if(osMutexAcquire(pubsub->mutex, osWaitForever) == osOK) {
+        bool result = osMutexDelete(pubsub->mutex) == osOK;
+        list_pubsub_cb_clear(pubsub->items);
+        return result;
+    } else {
+        return false;
+    }
+}
+
+PubSubItem* subscribe_pubsub(PubSub* pubsub, PubSubCallback cb, void* ctx) {
+    if(osMutexAcquire(pubsub->mutex, osWaitForever) == osOK) {
+        // put uninitialized item to the list
+        PubSubItem* item = list_pubsub_cb_push_raw(pubsub->items);
+
+        // initialize item
+        item->cb = cb;
+        item->ctx = ctx;
+        item->self = pubsub;
+
+        // TODO unsubscribe pubsub on app exit
+        //flapp_on_exit(unsubscribe_pubsub, item);
+
+        osMutexRelease(pubsub->mutex);
+
+        return item;
+    } else {
+        return NULL;
+    }
+}
+
+bool unsubscribe_pubsub(PubSubItem* pubsub_id) {
+    if(osMutexAcquire(pubsub_id->self->mutex, osWaitForever) == osOK) {
+        bool result = false;
+
+        // iterate over items
+        list_pubsub_cb_it_t it;
+        for(list_pubsub_cb_it(it, pubsub_id->self->items); !list_pubsub_cb_end_p(it);
+            list_pubsub_cb_next(it)) {
+            const PubSubItem* item = list_pubsub_cb_cref(it);
+
+            // if the iterator is equal to our element
+            if(item == pubsub_id) {
+                list_pubsub_cb_remove(pubsub_id->self->items, it);
+                result = true;
+                break;
+            }
+        }
+
+        osMutexRelease(pubsub_id->self->mutex);
+        return result;
+    } else {
+        return false;
+    }
+}
+
+bool notify_pubsub(PubSub* pubsub, void* arg) {
+    if(osMutexAcquire(pubsub->mutex, osWaitForever) == osOK) {
+        // iterate over subscribers
+        list_pubsub_cb_it_t it;
+        for(list_pubsub_cb_it(it, pubsub->items); !list_pubsub_cb_end_p(it);
+            list_pubsub_cb_next(it)) {
+            const PubSubItem* item = list_pubsub_cb_cref(it);
+            item->cb(arg, item->ctx);
+        }
+
+        osMutexRelease(pubsub->mutex);
+        return true;
+    } else {
+        return false;
+    }
+}

core/api-basic/pubsub.c.unimplemented

@@ -1,48 +0,0 @@
-#include "pubsub.h"
-
-void init_pubsub(PubSub* pubsub) {
-    pubsub->count = 0;
-
-    for(size_t i = 0; i < NUM_OF_CALLBACKS; i++) {
-        pubsub->items[i].
-    }
-}
-
-// TODO add mutex to reconfigurate PubSub
-PubSubId* subscribe_pubsub(PubSub* pubsub, PubSubCallback cb, void* ctx) {
-    if(pubsub->count >= NUM_OF_CALLBACKS) return NULL;
-
-    pubsub->count++;
-    PubSubItem* current = pubsub->items[pubsub->count];
-    
-    current->cb = cb;
-    currrnt->ctx = ctx;
-
-    pubsub->ids[pubsub->count].self = pubsub;
-    pubsub->ids[pubsub->count].item = current;
-
-    flapp_on_exit(unsubscribe_pubsub, &(pubsub->ids[pubsub->count]));
-    
-    return current;
-}
-
-void unsubscribe_pubsub(PubSubId* pubsub_id) {
-    // TODO: add, and rearrange all items to keep subscribers item continuous
-    // TODO: keep ids link actual
-    // TODO: also add mutex on every pubsub changes
-
-    // trivial implementation for NUM_OF_CALLBACKS = 1
-    if(NUM_OF_CALLBACKS != 1) return;
-
-    if(pubsub_id != NULL || pubsub_id->self != NULL || pubsub_id->item != NULL) return;
-
-    pubsub_id->self->count = 0;
-    pubsub_id->item = NULL;
-}
-
-void notify_pubsub(PubSub* pubsub, void* arg) {
-    // iterate over subscribers
-    for(size_t i = 0; i < pubsub->count; i++) {
-        pubsub->items[i]->cb(arg, pubsub->items[i]->ctx);
-    }
-}

core/api-basic/pubsub.h

@@ -1,6 +1,7 @@
 #pragma once
 
-#include "flipper.h"
+#include "flipper_v2.h"
+#include "m-list.h"
 
 /*
 == PubSub ==
@@ -11,43 +12,46 @@ and also subscriber can set `void*` context pointer that pass into
 callback (you can see callback signature below).
 */
 
-typedef void(PubSubCallback*)(void*, void*);
+typedef void (*PubSubCallback)(void*, void*);
+typedef struct PubSubType PubSub;
 
 typedef struct {
     PubSubCallback cb;
     void* ctx;
+    PubSub* self;
 } PubSubItem;
 
-typedef struct {
+LIST_DEF(list_pubsub_cb, PubSubItem, M_POD_OPLIST);
-    PubSub* self;
-    PubSubItem* item;
-} PubSubId;
 
-typedef struct {
+struct PubSubType {
-    PubSubItem items[NUM_OF_CALLBACKS];
+    list_pubsub_cb_t items;
-    PubSubId ids[NUM_OF_CALLBACKS]; ///< permanent links to item
+    osMutexId_t mutex;
-    size_t count; ///< count of callbacks
+};
-} PubSub;
 
 /*
 To create PubSub you should create PubSub instance and call `init_pubsub`.
 */
-void init_pubsub(PubSub* pubsub);
+bool init_pubsub(PubSub* pubsub);
+
+/*
+Since we use dynamic memory - we must explicity delete pubsub
+*/
+bool delete_pubsub(PubSub* pubsub);
 
 /*
 Use `subscribe_pubsub` to register your callback.
 */
-PubSubId* subscribe_pubsub(PubSub* pubsub, PubSubCallback cb, void* ctx);
+PubSubItem* subscribe_pubsub(PubSub* pubsub, PubSubCallback cb, void* ctx);
 
 /*
 Use `unsubscribe_pubsub` to unregister callback.
 */
-void unsubscribe_pubsub(PubSubId* pubsub_id);
+bool unsubscribe_pubsub(PubSubItem* pubsub_id);
 
 /*
 Use `notify_pubsub` to notify subscribers.
 */
-void notify_pubsub(PubSub* pubsub, void* arg);
+bool notify_pubsub(PubSub* pubsub, void* arg);
 
 /*
 

core/flipper_v2.h

@@ -4,6 +4,6 @@
 //#include "api-basic/flapp.h"
 #include "cmsis_os2.h"
 #include "api-basic/valuemutex.h"
-//#include "api-basic/pubsub.h"
+#include "api-basic/pubsub.h"
 
 #include "api-basic/memmgr.h"

firmware/targets/local/Inc/cmsis_os.h

@@ -94,5 +94,6 @@ typedef enum {
 
 osStatus_t osMutexAcquire (osMutexId_t mutex_id, uint32_t timeout);
 osStatus_t osMutexRelease (osMutexId_t mutex_id);
+osStatus_t osMutexDelete (osMutexId_t mutex_id);
 
 #define osWaitForever portMAX_DELAY

firmware/targets/local/Src/lo_os.c

@@ -253,3 +253,15 @@ osStatus_t osMutexRelease (osMutexId_t mutex_id) {
         return osError;
     }
 }
+
+osStatus_t osMutexDelete (osMutexId_t mutex_id) {
+    osMutexRelease(mutex_id);
+    
+    int res = 0;
+    if((res = pthread_mutex_destroy(&mutex_id->mutex)) == 0) {
+        return osOK;
+    } else {
+        printf("res = %d\n", res);
+        return osError;
+    }
+}