edi/buzzer_2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
01448223ad8e86cc6211e7b4004f7178a958f442
buzzer_2/firmware/libs/buzz_proto/src/buzz_proto.c
Eduard Iten 01448223ad zwischenstand
2026-03-21 13:49:05 +01:00

1044 lines
35 KiB
C
Raw Blame History

#include <zephyr/kernel.h>
#include <zephyr/logging/log.h>
#include <zephyr/logging/log_ctrl.h>
#include <zephyr/fs/fs.h>
#include <zephyr/sys/crc.h>
#include <errno.h>
#include <stdlib.h>
#include "buzz_proto.h"
#include "fs_mgmt.h"
#include "fw_mgmt.h"
LOG_MODULE_REGISTER(buzz_proto, CONFIG_BUZZ_PROTO_LOG_LEVEL);
K_MEM_SLAB_DEFINE(buzz_proto_slabs, CONFIG_BUZZ_PROTO_SLAB_SIZE, CONFIG_BUZZ_PROTO_SLAB_COUNT, 4);
K_MSGQ_DEFINE(buzz_proto_msgq, sizeof(struct buzz_frame_msg), CONFIG_BUZZ_PROTO_MSGQ_SIZE, 4);
struct ls_state_t
{
bool active;
int credits;
uint32_t entries_sent;
uint32_t retry_counter;
struct fs_dir_t dir;
struct fs_dirent entry;
buzz_transport_reply_fn reply_cb;
};
static struct ls_state_t ls_state = {
.active = false,
.credits = 0,
.entries_sent = 0,
.retry_counter = 0,
.reply_cb = NULL,
};
struct get_file_state_t
{
bool active;
int credits;
uint32_t offset;
uint32_t retry_counter;
uint32_t crc32;
struct fs_file_t file;
buzz_transport_reply_fn reply_cb;
uint16_t max_payload;
};
static struct get_file_state_t get_file_state = {
.active = false,
.credits = 0,
.offset = 0,
.retry_counter = 0,
.crc32 = 0,
.reply_cb = NULL,
};
enum stream_state_t
{
STREAM_IDLE,
STREAM_LS,
STREAM_FILE_PUT,
STREAM_FILE_GET,
STREAM_FW_UPDATE,
};
static enum stream_state_t current_stream = STREAM_IDLE;
static char src_path[FS_MGMT_MAX_PATH_LENGTH], dst_path[FS_MGMT_MAX_PATH_LENGTH];
int buzz_proto_buf_alloc(uint8_t **buf)
{
return k_mem_slab_alloc(&buzz_proto_slabs, (void **)buf, K_NO_WAIT);
}
void buzz_proto_buf_free(uint8_t **buf)
{
if (buf && *buf)
{
k_mem_slab_free(&buzz_proto_slabs, *buf);
*buf = NULL;
}
}
int buzz_proto_submit_frame(struct buzz_frame_msg *msg)
{
return k_msgq_put(&buzz_proto_msgq, msg, K_NO_WAIT);
}
static void send_stream_error(buzz_transport_reply_fn reply_cb, uint16_t error_code);
static void send_error_frame(struct buzz_frame_msg *msg, uint16_t error_code)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
struct buzz_resp_error *err = (struct buzz_resp_error *)(msg->data_ptr + sizeof(*hdr));
hdr->frame_type = BUZZ_FRAME_ERROR;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_error));
err->error_code = sys_cpu_to_le16(error_code);
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, sizeof(*hdr) + sizeof(*err));
}
}
static void send_stream_error(buzz_transport_reply_fn reply_cb, uint16_t error_code)
{
uint8_t *buf = NULL;
if (reply_cb == NULL || buzz_proto_buf_alloc(&buf) != 0)
{
return;
}
struct buzz_frame_msg err_msg = {.data_ptr = buf, .reply_cb = reply_cb};
send_error_frame(&err_msg, error_code);
buzz_proto_buf_free(&buf);
}
uint16_t buzz_proto_get_free_rx_slabs(void)
{
uint32_t free_slabs = k_mem_slab_num_free_get(&buzz_proto_slabs);
return (free_slabs > 4) ? (uint16_t)(free_slabs - 4) : 0;
}
void buzz_proto_send_ack(buzz_transport_reply_fn reply_cb, uint16_t credits)
{
if (!reply_cb || credits == 0)
return;
uint8_t *buf;
if (buzz_proto_buf_alloc(&buf) == 0)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)buf;
hdr->frame_type = BUZZ_FRAME_ACK;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_ack_payload));
struct buzz_ack_payload *pl = (struct buzz_ack_payload *)(buf + sizeof(*hdr));
pl->credits = sys_cpu_to_le16(credits);
reply_cb(buf, sizeof(*hdr) + sizeof(*pl));
buzz_proto_buf_free(&buf);
}
}
void buzz_proto_send_success_reusing_slab(buzz_transport_reply_fn reply_cb, uint8_t data_type, uint8_t *slab)
{
if (!reply_cb || !slab)
return;
struct buzz_proto_header *hdr = (struct buzz_proto_header *)slab;
hdr->frame_type = BUZZ_FRAME_SUCCESS;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_success));
struct buzz_resp_success *succ = (struct buzz_resp_success *)(slab + sizeof(*hdr));
succ->data_type = data_type;
reply_cb(slab, sizeof(*hdr) + sizeof(*succ));
buzz_proto_buf_free(&slab);
}
void buzz_proto_send_error_reusing_slab(buzz_transport_reply_fn reply_cb, uint16_t error_code, uint8_t *slab)
{
if (!reply_cb || !slab)
return;
struct buzz_proto_header *hdr = (struct buzz_proto_header *)slab;
hdr->frame_type = BUZZ_FRAME_ERROR;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_error));
struct buzz_resp_error *err = (struct buzz_resp_error *)(slab + sizeof(*hdr));
err->error_code = sys_cpu_to_le16(error_code);
reply_cb(slab, sizeof(*hdr) + sizeof(*err));
buzz_proto_buf_free(&slab);
}
static void handle_proto_version_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
hdr->frame_type = BUZZ_FRAME_RESPONSE;
struct buzz_resp_proto_version *resp_data = (struct buzz_resp_proto_version *)(msg->data_ptr + sizeof(*hdr));
resp_data->data_type = BUZZ_DATA_PROTO_INFO;
resp_data->version = sys_cpu_to_le16(BUZZ_PROTO_VERSION);
/* Dynamische Chunk-Grösse basierend auf der aktuellen Transport-MTU berechnen */
uint16_t slab_payload = CONFIG_BUZZ_PROTO_SLAB_SIZE - sizeof(struct buzz_proto_header);
uint16_t transport_payload = 0;
if (msg->max_payload > sizeof(struct buzz_proto_header)) {
transport_payload = msg->max_payload - sizeof(struct buzz_proto_header);
}
uint16_t safe_chunk = MIN(slab_payload, transport_payload);
resp_data->max_chunk_size = sys_cpu_to_le16(safe_chunk);
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_proto_version));
uint16_t total_len = sizeof(struct buzz_proto_header) + sizeof(struct buzz_resp_proto_version);
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, total_len);
}
}
void handle_device_info_request(struct buzz_frame_msg *msg) {
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
hdr->frame_type = BUZZ_FRAME_RESPONSE;
struct buzz_resp_device_info *resp_data = (struct buzz_resp_device_info *)(msg->data_ptr + sizeof(*hdr));
resp_data->data_type = BUZZ_DATA_DEVICE_INFO;
if (fw_mgmt_get_id(resp_data->device_id, sizeof(resp_data->device_id)) < 0) {
LOG_ERR("Failed to get device ID");
send_error_frame(msg, EIO);
return;
}
const char *board_name = fw_mgmt_get_board_name();
const char *board_rev = fw_mgmt_get_board_revision();
const char *soc_name = fw_mgmt_get_soc_name();
resp_data->board_name_length = MIN(strlen(board_name), 32); // Sicherheitsmassnahme gegen zu lange Namen
resp_data->board_revision_length = MIN(strlen(board_rev), 32); // Sicherheitsmassnahme gegen zu lange Namen
resp_data->soc_name_length = MIN(strlen(soc_name), 32); // Sicherheitsmassnahme gegen zu lange Namen
memcpy(resp_data->data, board_name, resp_data->board_name_length);
memcpy(resp_data->data + resp_data->board_name_length, board_rev, resp_data->board_revision_length);
memcpy(resp_data->data + resp_data->board_name_length + resp_data->board_revision_length, soc_name, resp_data->soc_name_length);
uint16_t payload_length = sizeof(struct buzz_resp_device_info) + resp_data->board_name_length + resp_data->board_revision_length + resp_data->soc_name_length;
hdr->payload_length = sys_cpu_to_le16(payload_length);
uint16_t total_len = sizeof(struct buzz_proto_header) + payload_length;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, total_len);
}
}
void handle_fs_info_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
struct fs_statvfs stat;
int rc = fs_mgmt_pm_statvfs(FS_AUDIO_PATH, &stat);
if (rc != 0)
{
LOG_ERR("Failed to statvfs audio path");
send_error_frame(msg, abs(rc));
return;
}
hdr->frame_type = BUZZ_FRAME_RESPONSE;
struct buzz_resp_fs_info *resp_data = (struct buzz_resp_fs_info *)(msg->data_ptr + sizeof(*hdr));
uint32_t block_size = stat.f_frsize;
uint32_t total_size = stat.f_blocks * block_size;
uint32_t free_size = stat.f_bfree * block_size;
LOG_DBG("FS Info: block_size=%u, total_size=%u, free_size=%u", block_size, total_size, free_size);
resp_data->data_type = BUZZ_DATA_FS_INFO;
resp_data->total_size = sys_cpu_to_le32(total_size);
resp_data->free_size = sys_cpu_to_le32(free_size);
resp_data->max_path_length = FS_MGMT_MAX_PATH_LENGTH;
resp_data->sys_path_length = strlen(FS_SYSTEM_PATH);
resp_data->audio_path_length = strlen(FS_AUDIO_PATH);
memcpy(resp_data->data, FS_SYSTEM_PATH, resp_data->sys_path_length);
memcpy(resp_data->data + resp_data->sys_path_length, FS_AUDIO_PATH, resp_data->audio_path_length);
uint16_t payload_length = sizeof(struct buzz_resp_fs_info) + resp_data->sys_path_length + resp_data->audio_path_length;
hdr->payload_length = sys_cpu_to_le16(payload_length);
uint16_t total_len = sizeof(struct buzz_proto_header) + payload_length;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, total_len);
}
}
static void handle_fw_info_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
hdr->frame_type = BUZZ_FRAME_RESPONSE;
struct buzz_resp_fw_info *resp_data = (struct buzz_resp_fw_info *)(msg->data_ptr + sizeof(*hdr));
resp_data->data_type = BUZZ_DATA_FW_INFO;
resp_data->fw_status = fw_mgmt_get_fw_state();
resp_data->slot1_size = sys_cpu_to_le32(fw_mgmt_get_slot1_size());
const char *fw_version = fw_mgmt_get_fw_version_string();
const char *kernel_version = fw_mgmt_get_kernel_version_string();
resp_data->fw_version_length = MIN(strlen(fw_version), 32); // Sicherheitsmassnahme gegen zu lange Strings
resp_data->kernel_version_length = MIN(strlen(kernel_version), 32); // Sicherheitsmassnahme gegen zu lange Strings
memcpy(resp_data->data, fw_version, resp_data->fw_version_length);
memcpy(resp_data->data + resp_data->fw_version_length, kernel_version, resp_data->kernel_version_length);
uint16_t payload_length = sizeof(struct buzz_resp_fw_info) + resp_data->fw_version_length + resp_data->kernel_version_length;
hdr->payload_length = sys_cpu_to_le16(payload_length);
uint16_t total_len = sizeof(struct buzz_proto_header) + payload_length;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, total_len);
}
}
static void handle_ls_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
uint16_t payload_len = sys_le16_to_cpu(hdr->payload_length);
if (current_stream != STREAM_IDLE)
{
LOG_WRN("Stream active, rejecting LS request");
send_error_frame(msg, EBUSY);
return;
}
uint16_t path_len = payload_len - 1;
if (path_len >= sizeof(src_path))
{
LOG_ERR("Path too long for LS request");
send_error_frame(msg, ENAMETOOLONG);
return;
}
memcpy(src_path, msg->data_ptr + sizeof(*hdr) + 1, path_len);
src_path[path_len] = '\0';
int rc = fs_mgmt_pm_opendir(&ls_state.dir, src_path);
if (rc != 0)
{
LOG_ERR("Failed to open dir: %d", rc);
send_error_frame(msg, abs(rc));
return;
}
current_stream = STREAM_LS;
ls_state.active = true;
ls_state.credits = 0;
ls_state.entries_sent = 0;
ls_state.retry_counter = 0;
ls_state.reply_cb = msg->reply_cb;
LOG_DBG("Started LS stream for path '%s'", src_path);
hdr->frame_type = BUZZ_FRAME_LS_START;
hdr->payload_length = 0;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, sizeof(*hdr));
}
}
static void handle_file_get_request(struct buzz_frame_msg *msg, bool only_tags)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
uint16_t payload_len = sys_le16_to_cpu(hdr->payload_length);
if (current_stream != STREAM_IDLE)
{
LOG_WRN("Stream active, rejecting FILE_GET request");
send_error_frame(msg, EBUSY);
return;
}
uint16_t path_len = payload_len - 1; // 1 Byte für data_type abziehen
if (path_len >= sizeof(src_path))
{
LOG_ERR("Path too long for FILE_GET request");
send_error_frame(msg, ENAMETOOLONG);
return;
}
memcpy(src_path, msg->data_ptr + sizeof(*hdr) + 1, path_len);
src_path[path_len] = '\0';
struct fs_dirent entry;
if (fs_mgmt_pm_stat(src_path, &entry) != 0)
{
LOG_ERR("File not found: %s", src_path);
send_error_frame(msg, ENOENT);
return;
}
fs_file_t_init(&get_file_state.file);
int rc = fs_mgmt_pm_open(&get_file_state.file, src_path, FS_O_READ);
if (rc != 0)
{
LOG_ERR("Failed to open file: %d", rc);
send_error_frame(msg, abs(rc));
return;
}
uint32_t stream_size = entry.size;
if (only_tags)
{
ssize_t audio_len = fs_get_audio_data_len(&get_file_state.file);
if (audio_len < 0)
{
LOG_ERR("Failed to get audio data len: %d", (int)audio_len);
fs_mgmt_pm_close(&get_file_state.file);
send_error_frame(msg, EIO);
return;
}
stream_size = entry.size - audio_len;
if (stream_size == 0)
{
fs_seek(&get_file_state.file, entry.size, FS_SEEK_SET);
}
else
{
fs_seek(&get_file_state.file, audio_len, FS_SEEK_SET);
}
}
current_stream = STREAM_FILE_GET;
get_file_state.active = true;
get_file_state.credits = 0;
get_file_state.offset = 0;
get_file_state.crc32 = 0; // IEEE CRC32 Startwert
get_file_state.retry_counter = 0;
get_file_state.reply_cb = msg->reply_cb;
get_file_state.max_payload = msg->max_payload;
LOG_INF("Started FILE_GET stream for '%s' (%u bytes)", src_path, entry.size);
hdr->frame_type = BUZZ_FRAME_FILE_START;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_file_start_payload));
struct buzz_file_start_payload *start_pl = (struct buzz_file_start_payload *)(msg->data_ptr + sizeof(*hdr));
start_pl->total_size = sys_cpu_to_le32(stream_size);
if (msg->reply_cb)
{
int send_rc = msg->reply_cb(msg->data_ptr, sizeof(*hdr) + sizeof(*start_pl));
if (send_rc)
{
LOG_ERR("Failed to send FILE_START (err %d)", send_rc);
fs_mgmt_pm_close(&get_file_state.file);
get_file_state.active = false;
current_stream = STREAM_IDLE;
k_sleep(K_MSEC(10));
send_error_frame(msg, EIO);
return;
}
}
}
static void process_rm_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
uint16_t payload_len = sys_le16_to_cpu(hdr->payload_length);
struct buzz_rm_file_payload *req = (struct buzz_rm_file_payload *)(msg->data_ptr + sizeof(*hdr));
if (payload_len < (sizeof(req->data_type) + sizeof(req->path_length)))
{
LOG_ERR("Invalid payload for RM_FILE request");
send_error_frame(msg, EINVAL);
return;
}
if ((sizeof(req->data_type) + sizeof(req->path_length) + req->path_length) != payload_len)
{
LOG_ERR("Path length in RM_FILE does not match payload length");
send_error_frame(msg, EINVAL);
return;
}
if (req->path_length >= sizeof(src_path))
{
LOG_ERR("Path too long for RM_FILE request");
send_error_frame(msg, ENAMETOOLONG);
return;
}
memcpy(src_path, req->path, req->path_length);
src_path[req->path_length] = '\0';
int rc = fs_mgmt_pm_unlink(src_path);
if (rc != 0)
{
LOG_ERR("Failed to remove file '%s': %d", src_path, rc);
send_error_frame(msg, abs(rc));
return;
}
LOG_INF("File '%s' removed successfully", src_path);
hdr->frame_type = BUZZ_FRAME_SUCCESS;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_success));
struct buzz_resp_success *resp_data = (struct buzz_resp_success *)(msg->data_ptr + sizeof(*hdr));
resp_data->data_type = BUZZ_DATA_RM_FILE;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, sizeof(*hdr) + sizeof(*resp_data));
}
}
static void process_move_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
uint16_t payload_len = sys_le16_to_cpu(hdr->payload_length);
struct buzz_rename_file_payload *req = (struct buzz_rename_file_payload *)(msg->data_ptr + sizeof(*hdr));
if (payload_len < (sizeof(req->data_type) + sizeof(req->old_path_length) + sizeof(req->new_path_length)))
{
LOG_ERR("Invalid payload for RENAME_FILE request");
send_error_frame(msg, EINVAL);
return;
}
if ((sizeof(req->data_type) + sizeof(req->old_path_length) + sizeof(req->new_path_length) + req->old_path_length + req->new_path_length) != payload_len)
{
LOG_ERR("Path lengths in RENAME_FILE do not match payload length");
send_error_frame(msg, EINVAL);
return;
}
if (req->old_path_length >= sizeof(src_path) || req->new_path_length >= sizeof(dst_path))
{
LOG_ERR("Source or destination path too long for RENAME_FILE request");
send_error_frame(msg, ENAMETOOLONG);
return;
}
memcpy(src_path, req->paths, req->old_path_length);
src_path[req->old_path_length] = '\0';
memcpy(dst_path, req->paths + req->old_path_length, req->new_path_length);
dst_path[req->new_path_length] = '\0';
int rc = fs_mgmt_pm_rename(src_path, dst_path);
if (rc != 0)
{
LOG_ERR("Failed to rename file from '%s' to '%s': %d", src_path, dst_path, rc);
send_error_frame(msg, abs(rc));
return;
}
LOG_INF("File renamed from '%s' to '%s' successfully", src_path, dst_path);
hdr->frame_type = BUZZ_FRAME_SUCCESS;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_resp_success));
struct buzz_resp_success *resp_data = (struct buzz_resp_success *)(msg->data_ptr + sizeof(*hdr));
resp_data->data_type = BUZZ_DATA_RENAME_FILE;
if (msg->reply_cb)
{
msg->reply_cb(msg->data_ptr, sizeof(*hdr) + sizeof(*resp_data));
}
}
static void process_file_get_stream(void)
{
uint8_t *buf = NULL;
if (buzz_proto_buf_alloc(&buf) != 0)
{
return; // Puffer voll, im nächsten Zyklus nochmal probieren
}
struct buzz_proto_header *hdr = (struct buzz_proto_header *)buf;
uint8_t *payload_ptr = buf + sizeof(*hdr);
if (get_file_state.max_payload <= sizeof(*hdr))
{
struct buzz_frame_msg err_msg = {
.data_ptr = buf,
.reply_cb = get_file_state.reply_cb,
.max_payload = get_file_state.max_payload,
};
send_error_frame(&err_msg, EMSGSIZE);
fs_mgmt_pm_close(&get_file_state.file);
get_file_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
LOG_ERR("Invalid max payload for FILE_GET: %u", get_file_state.max_payload);
return;
}
// Chunk Size berechnen
uint16_t max_chunk_size = MIN(
get_file_state.max_payload - sizeof(*hdr),
CONFIG_BUZZ_PROTO_SLAB_SIZE - sizeof(struct buzz_proto_header));
ssize_t read_len = fs_read(&get_file_state.file, payload_ptr, max_chunk_size);
if (read_len < 0)
{
// Lesefehler
struct buzz_frame_msg err_msg = {.data_ptr = buf, .reply_cb = get_file_state.reply_cb, .max_payload = get_file_state.max_payload};
send_error_frame(&err_msg, EIO);
fs_mgmt_pm_close(&get_file_state.file);
get_file_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
LOG_ERR("Error reading file: %d", (int)read_len);
return;
}
if (read_len == 0)
{
// EOF erreicht -> FILE_END senden
hdr->frame_type = BUZZ_FRAME_FILE_END;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_file_end_payload));
struct buzz_file_end_payload *end_pl = (struct buzz_file_end_payload *)payload_ptr;
end_pl->crc32 = sys_cpu_to_le32(get_file_state.crc32);
if (get_file_state.reply_cb)
{
int send_rc = get_file_state.reply_cb(buf, sizeof(*hdr) + sizeof(*end_pl));
if (send_rc)
{
LOG_WRN("Failed to send FILE_END (err %d)", send_rc);
}
}
fs_mgmt_pm_close(&get_file_state.file);
get_file_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
LOG_INF("FILE_GET stream ended. CRC32: 0x%08X", get_file_state.crc32);
return;
}
// Chunk senden; CRC/Offset erst nach erfolgreichem Enqueue aktualisieren
hdr->frame_type = BUZZ_FRAME_FILE_CHUNK;
hdr->payload_length = sys_cpu_to_le16(read_len);
if (get_file_state.reply_cb)
{
int send_rc = get_file_state.reply_cb(buf, sizeof(*hdr) + read_len);
if (send_rc == -ENOMEM)
{
// BLE TX queue voll - Datei zurücksetzen, nächster Zyklus wiederholt den Chunk
fs_seek(&get_file_state.file, -(off_t)read_len, FS_SEEK_CUR);
buzz_proto_buf_free(&buf);
return;
}
if (send_rc)
{
LOG_ERR("Failed to send FILE_CHUNK (err %d)", send_rc);
fs_mgmt_pm_close(&get_file_state.file);
get_file_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
k_sleep(K_MSEC(10));
send_stream_error(get_file_state.reply_cb, EIO);
return;
}
}
// Erfolgreich eingereiht: State aktualisieren
get_file_state.crc32 = crc32_ieee_update(get_file_state.crc32, payload_ptr, read_len);
get_file_state.offset += read_len;
get_file_state.credits--;
get_file_state.retry_counter = 0;
buzz_proto_buf_free(&buf);
}
static void handle_request(struct buzz_frame_msg *msg)
{
struct buzz_proto_header *hdr = (struct buzz_proto_header *)msg->data_ptr;
uint16_t payload_len = sys_le16_to_cpu(hdr->payload_length);
if (payload_len < sizeof(struct buzz_request_payload))
{
LOG_WRN("Invalid request length");
send_error_frame(msg, EINVAL);
return;
}
struct buzz_request_payload *req_data = (struct buzz_request_payload *)(msg->data_ptr + sizeof(*hdr));
switch (req_data->data_type)
{
case BUZZ_DATA_PROTO_INFO:
LOG_DBG("Received Proto Version Request");
handle_proto_version_request(msg);
break;
case BUZZ_DATA_DEVICE_INFO:
LOG_DBG("Received Device Info Request");
handle_device_info_request(msg);
break;
case BUZZ_DATA_FS_INFO:
LOG_DBG("Received FS Info Request");
handle_fs_info_request(msg);
break;
case BUZZ_DATA_LS:
LOG_DBG("Received LS Request");
handle_ls_request(msg);
break;
case BUZZ_DATA_FW_INFO:
LOG_DBG("Received FW Info Request");
handle_fw_info_request(msg);
break;
case BUZZ_DATA_FILE_GET:
LOG_DBG("Received FILE_GET Request");
handle_file_get_request(msg, false);
break;
case BUZZ_DATA_FILE_PUT:
LOG_DBG("Received FILE_PUT Request");
if (payload_len < sizeof(struct buzz_request_payload) + sizeof(uint32_t) + 1)
{
send_error_frame(msg, EINVAL);
return;
}
if (current_stream != STREAM_IDLE)
{
LOG_WRN("Stream active, rejecting FILE_PUT request");
send_error_frame(msg, EBUSY);
return;
}
struct fs_write_msg write_req = {
.op = FS_WRITE_OP_FILE_START,
.slab_ptr = msg->data_ptr,
.data_offset = sizeof(*hdr) + sizeof(struct buzz_request_payload) + sizeof(uint32_t),
.data_len = payload_len - sizeof(struct buzz_request_payload) - sizeof(uint32_t),
.metadata = sys_get_le32(msg->data_ptr + sizeof(*hdr) + sizeof(struct buzz_request_payload)),
.reply_cb = msg->reply_cb};
if (fs_mgmt_submit_write(&write_req) == 0)
{
current_stream = STREAM_FILE_PUT; /* WICHTIG: Status blockieren */
msg->data_ptr = NULL; /* Ownership an FS-Thread übertragen */
}
else
{
send_error_frame(msg, EBUSY);
}
break;
case BUZZ_DATA_TAGS_PUT:
LOG_DBG("Received TAGS_PUT Request");
if (payload_len < sizeof(struct buzz_request_payload) + sizeof(uint32_t) + 1)
{
send_error_frame(msg, EINVAL);
return;
}
if (current_stream != STREAM_IDLE)
{
LOG_WRN("Stream active, rejecting TAGS_PUT request");
send_error_frame(msg, EBUSY);
return;
}
struct fs_write_msg tags_req = {
.op = FS_WRITE_OP_TAGS_START,
.slab_ptr = msg->data_ptr,
.data_offset = sizeof(*hdr) + sizeof(struct buzz_request_payload) + sizeof(uint32_t),
.data_len = payload_len - sizeof(struct buzz_request_payload) - sizeof(uint32_t),
.metadata = sys_get_le32(msg->data_ptr + sizeof(*hdr) + sizeof(struct buzz_request_payload)),
.reply_cb = msg->reply_cb
};
if (fs_mgmt_submit_write(&tags_req) == 0)
{
current_stream = STREAM_FILE_PUT; /* Blockiert den Stream für weitere Requests */
msg->data_ptr = NULL;
}
else
{
send_error_frame(msg, EBUSY);
}
break;
case BUZZ_DATA_TAGS_GET:
LOG_DBG("Received TAGS_GET Request");
handle_file_get_request(msg, true);
break;
case BUZZ_DATA_RM_FILE:
LOG_DBG("Received RM_FILE Request");
process_rm_request(msg);
break;
case BUZZ_DATA_RENAME_FILE:
LOG_DBG("Received RENAME_FILE Request");
process_move_request(msg);
break;
default:
LOG_WRN("Unknown request data_type: 0x%02x", req_data->data_type);
send_error_frame(msg, EINVAL);
break;
}
}
static void process_ls_stream(void)
{
uint8_t *buf = NULL;
if (buzz_proto_buf_alloc(&buf) != 0)
{
return; // Nächster Versuch im nächsten Zyklus
}
int rc = fs_readdir(&ls_state.dir, &ls_state.entry);
if (rc < 0)
{
struct buzz_frame_msg err_msg = {.data_ptr = buf, .reply_cb = ls_state.reply_cb};
send_error_frame(&err_msg, abs(rc));
fs_mgmt_pm_closedir(&ls_state.dir);
ls_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
LOG_ERR("Error reading directory: %d", rc);
return;
}
struct buzz_proto_header *hdr = (struct buzz_proto_header *)buf;
if (ls_state.entry.name[0] == 0)
{
hdr->frame_type = BUZZ_FRAME_LS_END;
hdr->payload_length = sys_cpu_to_le16(sizeof(struct buzz_ls_end_payload));
struct buzz_ls_end_payload *end_pl = (struct buzz_ls_end_payload *)(buf + sizeof(*hdr));
end_pl->total_entries = sys_cpu_to_le32(ls_state.entries_sent);
if (ls_state.reply_cb)
{
ls_state.reply_cb(buf, sizeof(*hdr) + sizeof(*end_pl));
}
fs_mgmt_pm_closedir(&ls_state.dir);
ls_state.active = false;
current_stream = STREAM_IDLE;
buzz_proto_buf_free(&buf);
LOG_DBG("LS stream ended. Total entries sent: %u", ls_state.entries_sent);
return;
}
hdr->frame_type = BUZZ_FRAME_LS_ENTRY;
struct buzz_ls_entry_payload *entry_pl = (struct buzz_ls_entry_payload *)(buf + sizeof(*hdr));
entry_pl->type = (ls_state.entry.type == FS_DIR_ENTRY_DIR) ? BUZZ_FS_ENTRY_DIR : BUZZ_FS_ENTRY_FILE;
entry_pl->size = sys_cpu_to_le32(ls_state.entry.size);
size_t name_len = strlen(ls_state.entry.name);
entry_pl->name_length = (uint8_t)name_len;
memcpy(entry_pl->name, ls_state.entry.name, name_len);
uint16_t payload_len = sizeof(*entry_pl) + name_len;
hdr->payload_length = sys_cpu_to_le16(payload_len);
if (ls_state.reply_cb)
{
ls_state.reply_cb(buf, sizeof(*hdr) + payload_len);
}
ls_state.credits--;
ls_state.entries_sent++;
ls_state.retry_counter = 0;
buzz_proto_buf_free(&buf);
}
static void buzz_proto_thread_fn(void *p1, void *p2, void *p3)
{
struct buzz_frame_msg msg;
struct buzz_proto_header *hdr;
LOG_INF("Buzz Protocol Thread started");
while (1)
{
k_timeout_t wait_time = K_FOREVER;
if ((current_stream == STREAM_LS && ls_state.active && ls_state.credits > 0) ||
(current_stream == STREAM_FILE_GET && get_file_state.active && get_file_state.credits > 0))
{
wait_time = K_NO_WAIT;
}
else if (current_stream != STREAM_IDLE)
{
wait_time = K_MSEC(500); // Watchdog Timeout
}
int q_res = k_msgq_get(&buzz_proto_msgq, &msg, wait_time);
/* 1. Eingehende Nachrichten verarbeiten */
if (q_res == 0)
{
if (msg.length < sizeof(struct buzz_proto_header))
{
LOG_WRN("Received frame too short");
buzz_proto_buf_free(&msg.data_ptr);
continue;
}
hdr = (struct buzz_proto_header *)msg.data_ptr;
switch (hdr->frame_type)
{
case BUZZ_FRAME_REQUEST:
handle_request(&msg);
buzz_proto_buf_free(&msg.data_ptr);
break;
case BUZZ_FRAME_ACK:
if (current_stream == STREAM_LS && msg.length >= sizeof(*hdr) + sizeof(struct buzz_ack_payload))
{
struct buzz_ack_payload *ack = (struct buzz_ack_payload *)(msg.data_ptr + sizeof(*hdr));
// Absolute Credits übernehmen, wie von dir vorgeschlagen
ls_state.credits = sys_le16_to_cpu(ack->credits);
ls_state.retry_counter = 0;
LOG_DBG("Got %u credits", ls_state.credits);
}
else if (current_stream == STREAM_FILE_GET && msg.length >= sizeof(*hdr) + sizeof(struct buzz_ack_payload))
{
struct buzz_ack_payload *ack = (struct buzz_ack_payload *)(msg.data_ptr + sizeof(*hdr));
get_file_state.credits = sys_le16_to_cpu(ack->credits);
get_file_state.retry_counter = 0;
}
buzz_proto_buf_free(&msg.data_ptr);
break;
case BUZZ_FRAME_FILE_CHUNK:
if (current_stream != STREAM_FILE_PUT)
{
send_error_frame(&msg, EBADMSG);
buzz_proto_buf_free(&msg.data_ptr);
break;
}
struct fs_write_msg chunk_req = {
.op = FS_WRITE_OP_FILE_CHUNK,
.slab_ptr = msg.data_ptr,
.data_offset = sizeof(*hdr),
.data_len = sys_le16_to_cpu(hdr->payload_length),
.reply_cb = msg.reply_cb};
if (fs_mgmt_submit_write(&chunk_req) == 0)
{
msg.data_ptr = NULL;
}
else
{
send_error_frame(&msg, EBUSY);
}
buzz_proto_buf_free(&msg.data_ptr); /* Tut nichts, wenn msg.data_ptr == NULL */
break;
case BUZZ_FRAME_FILE_END:
if (current_stream != STREAM_FILE_PUT)
{
send_error_frame(&msg, EBADMSG);
buzz_proto_buf_free(&msg.data_ptr);
break;
}
if (msg.length >= sizeof(*hdr) + sizeof(struct buzz_file_end_payload))
{
struct buzz_file_end_payload *end_pl = (struct buzz_file_end_payload *)(msg.data_ptr + sizeof(*hdr));
struct fs_write_msg end_req = {
.op = FS_WRITE_OP_FILE_END,
.slab_ptr = msg.data_ptr,
.data_offset = 0,
.data_len = 0,
.metadata = sys_le32_to_cpu(end_pl->crc32),
.reply_cb = msg.reply_cb};
if (fs_mgmt_submit_write(&end_req) == 0)
{
msg.data_ptr = NULL;
current_stream = STREAM_IDLE; /* Stream wieder freigeben */
}
else
{
send_error_frame(&msg, EBUSY);
}
}
else
{
send_error_frame(&msg, EINVAL);
}
buzz_proto_buf_free(&msg.data_ptr);
break;
case BUZZ_FRAME_FW_CHUNK:
send_error_frame(&msg, ENOSYS);
buzz_proto_buf_free(&msg.data_ptr);
break;
case BUZZ_FRAME_LS_ENTRY:
send_error_frame(&msg, ENOSYS);
buzz_proto_buf_free(&msg.data_ptr);
break;
default:
LOG_WRN("Unhandled frame type: 0x%02x", hdr->frame_type);
send_error_frame(&msg, EPROTO);
buzz_proto_buf_free(&msg.data_ptr);
break;
}
}
if (current_stream == STREAM_LS && ls_state.active)
{
if (ls_state.credits > 0)
{
process_ls_stream();
}
else if (q_res == -EAGAIN)
{
// Watchdog: Queue hat 500ms blockiert, weil keine Credits (ACK) kamen
ls_state.retry_counter++;
if (ls_state.retry_counter > 5)
{
LOG_WRN("LS timeout waiting for ACK");
fs_mgmt_pm_closedir(&ls_state.dir);
send_stream_error(ls_state.reply_cb, ETIMEDOUT);
ls_state.active = false;
current_stream = STREAM_IDLE;
}
}
}
else if (current_stream == STREAM_FILE_GET && get_file_state.active)
{
if (get_file_state.credits > 0)
{
process_file_get_stream();
}
else if (q_res == -EAGAIN)
{
get_file_state.retry_counter++;
if (get_file_state.retry_counter > 5)
{
LOG_WRN("FILE_GET timeout waiting for ACK");
fs_close(&get_file_state.file);
send_stream_error(get_file_state.reply_cb, ETIMEDOUT);
get_file_state.active = false;
current_stream = STREAM_IDLE;
}
}
}
}
}
K_THREAD_DEFINE(buzz_proto_thread, CONFIG_BUZZ_PROT_THREAD_STACK_SIZE, buzz_proto_thread_fn, NULL, NULL, NULL, CONFIG_BUZZ_PROTO_THREAD_PRIORITY, 0, 0);
Reference in New Issue View Git Blame Copy Permalink