diff --git a/app/src/incphub/src/incphub_client.c b/app/src/incphub/src/incphub_client.c index e135c97..fac56b9 100644 --- a/app/src/incphub/src/incphub_client.c +++ b/app/src/incphub/src/incphub_client.c @@ -80,7 +80,7 @@ int incphub_send_local(incphub_cli_t *cli, uint8_t *buffer, size_t length, ntl_p nil_build_header(&nil_packet->header, NIL_ADDR_LOCALHOST, INCPHUB_LOCAL_NIL_ADDR, length + sizeof(ntl_header_t)); // Pass to ingest - return enqueue_message_ingest((uint8_t *)nil_packet, extended_len); + return enqueue_message_ingest((uint8_t *)nil_packet, extended_len, portMAX_DELAY); } int incphub_await_msg(incphub_cli_t *cli, uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr) diff --git a/app/src/incphub/src/incphub_shared.c b/app/src/incphub/src/incphub_shared.c index 58ba719..f39a565 100644 --- a/app/src/incphub/src/incphub_shared.c +++ b/app/src/incphub/src/incphub_shared.c @@ -172,18 +172,55 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr) return INCPHUB_OK; } -int enqueue_message_ingest(uint8_t *buffer, size_t length) +int enqueue_message_ingest(uint8_t *buffer, size_t length, TickType_t max_delay) { - (void)buffer; - (void)length; + //NOTE: Does not need global lock + + if (buffer == NULL) + return INCPHUB_EINVAL; + + // Build message definition + incphub_msg_t msg = { .buffer = buffer, .length = length }; + + // Try to queue message + if (xQueueSendToBack(incphub_ingest_queue, &msg, max_delay) != pdPASS) + { + return INCPHUB_EAGAIN; + } + + return INCPHUB_OK; +} + +int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr, TickType_t max_delay) +{ + //NOTE: Does not need global lock + + (void)buffer_ptr; + (void)length_ptr; + (void)max_delay; //TODO: Implement return INCPHUB_ERR; } -int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr) +int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source) { + //NOTE: Does not need global lock + + (void)cli; + (void)buffer; + (void)length; + (void)source; + //TODO: Implement + return INCPHUB_ERR; +} + +int await_message_dispatch(uint8_t **buffer_ptr, size_t *length_ptr, incphub_addr_t *source_ptr) +{ + //NOTE: Does not need global lock + (void)buffer_ptr; (void)length_ptr; + (void)source_ptr; //TODO: Implement return INCPHUB_ERR; } diff --git a/app/src/incphub/src/incphub_shared.h b/app/src/incphub/src/incphub_shared.h index 6b0a72a..22d5394 100644 --- a/app/src/incphub/src/incphub_shared.h +++ b/app/src/incphub/src/incphub_shared.h @@ -20,6 +20,11 @@ struct incphub_cli_t { TaskHandle_t task; }; +typedef struct { + uint8_t *buffer; + size_t length; +} incphub_msg_t; + // Queue extern QueueHandle_t incphub_ingest_queue; @@ -60,14 +65,16 @@ int get_client_with_port(ntl_port_t port, incphub_cli_t **client_ptr); /// @brief Enqueues a message for ingest by the incphub task /// @param buffer The buffer to enqueue /// @param length The message length +/// @param max_delay The maximum time to wait for resources /// @return INCPHUB_OK on success, an error otherwise -int enqueue_message_ingest(uint8_t *buffer, size_t length); +int enqueue_message_ingest(uint8_t *buffer, size_t length, TickType_t max_delay); /// @brief Dequeues a message for ingest /// @param buffer_ptr Location to return the buffer /// @param length_ptr Location to store the message length +/// @param max_delay The maximum time to wait for resources /// @return INCPHUB_OK on success, an error otherwise -int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr); +int dequeue_message_ingest(uint8_t **buffer_ptr, size_t *length_ptr, TickType_t maxDelay); int do_message_dispatch(incphub_cli_t *cli, uint8_t *buffer, size_t length, incphub_addr_t source); diff --git a/app/src/incphub/src/incphub_task.c b/app/src/incphub/src/incphub_task.c index 83f4c5a..d0de7b3 100644 --- a/app/src/incphub/src/incphub_task.c +++ b/app/src/incphub/src/incphub_task.c @@ -29,7 +29,7 @@ QueueHandle_t incphub_ingest_queue; // === Private variables === #define INGEST_QUEUE_LENGTH (16) -#define INGEST_QUEUE_ITEM_SIZE (sizeof(ntl_packet_t *)) +#define INGEST_QUEUE_ITEM_SIZE (sizeof(incphub_msg_t)) static uint8_t _ingest_queue_store[INGEST_QUEUE_LENGTH * INGEST_QUEUE_ITEM_SIZE]; static StaticQueue_t _ingest_queue_static; @@ -60,7 +60,7 @@ void incphub_task_main(void *params) // Dequeue message uint8_t *buffer; size_t length; - dequeue_message_ingest(&buffer, &length); + dequeue_message_ingest(&buffer, &length, portMAX_DELAY); // Get NIL packet nil_packet_t *nil = extract_nil_packet(buffer, length);