// SPDX-License-Identifier: BSD-3-Clause OR Apache-2.0 #include "agents/sqlite_agent.h" #include "cla/cla.h" #include "cla/cla_contact_tx_task.h" #include "cla/posix/cla_sqlite.h" #include "platform/hal_io.h" #include "platform/hal_queue.h" #include "platform/hal_semaphore.h" #include "platform/hal_task.h" #include "platform/hal_types.h" #include "platform/posix/hal_types.h" #include "ud3tn/eid.h" #include "ud3tn/result.h" #include "ud3tn/simplehtab.h" #include #include #include #include #include static const char *CLA_NAME = "sqlite"; static const char *SQL_CREATE_TABLE = ( "CREATE TABLE IF NOT EXISTS bundles(source TEXT NOT NULL, destination TEXT, creation_timestamp INTEGER NOT NULL, bundle BLOB, PRIMARY KEY(source, creation_timestamp));" ); static const char *SQL_INSERT_INTO = ( "INSERT INTO bundles(source, destination, creation_timestamp, bundle) VALUES (?, ?, ?, ?);" ); struct sqlite_bundle_blob { sqlite3_blob *blob; sqlite3_int64 size; sqlite3_int64 offset; }; static enum ud3tn_result sqlite_bundle_blob_open( struct sqlite3 *db, sqlite3_int64 row_id, struct sqlite_bundle_blob *bundle_blob) { if (sqlite3_blob_open( db, "main", "bundles", "bundle", row_id, 1, &bundle_blob->blob ) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to open BLOB: %s", sqlite3_errmsg(db)); return UD3TN_FAIL; } bundle_blob->size = sqlite3_blob_bytes(bundle_blob->blob); bundle_blob->offset = 0; return UD3TN_OK; } static void sqlite_bundle_blob_reset(struct sqlite_bundle_blob *bundle_blob) { bundle_blob->blob = NULL; bundle_blob->size = 0; bundle_blob->offset = 0; } static enum ud3tn_result sqlite_bundle_blob_close( struct sqlite3 *db, struct sqlite_bundle_blob *bundle_blob) { int32_t res = sqlite3_blob_close(bundle_blob->blob); if (res != SQLITE_OK) LOGF_ERROR("SQLiteCLA: Failed to close BLOB: %s", sqlite3_errmsg(db)); sqlite_bundle_blob_reset(bundle_blob); return res != SQLITE_OK ? UD3TN_FAIL : UD3TN_OK; } struct sqlite_link { struct cla_link base; struct sqlite3 *db; QueueIdentifier_t agent_queue; struct sqlite_bundle_blob read_bundle_blob; struct sqlite_bundle_blob send_bundle_blob; }; struct sqlite_config { struct cla_config base; struct sqlite_link *link; pthread_t connection_task; }; const char *sqlite_name_get(void) { return CLA_NAME; } static enum ud3tn_result sqlite_launch(struct cla_config *const config) { return UD3TN_OK; } static void sqlite_disconnect_handler(struct cla_link *link) { // UNUSED (void)link; } static enum ud3tn_result sqlite_terminate(struct cla_config *config) { struct sqlite_config *sqlite_config = (struct sqlite_config *)config; // Signal the termination of RX/TX task cla_generic_disconnect_handler((struct cla_link *)sqlite_config->link); // Signal the termination of sqlite_read() struct cla_sqlite_command cmd = { .type = SQLITE_COMMAND_FINALIZE }; hal_queue_push_to_back(sqlite_config->link->agent_queue, &cmd); // Close database connection sqlite3_close(sqlite_config->link->db); // Cleanup allocated resources cla_link_wait_cleanup(&sqlite_config->link->base); hal_queue_delete(sqlite_config->link->agent_queue); free(sqlite_config->link); free(sqlite_config); LOG_INFO("SQLiteCLA: Terminated gracefully"); return UD3TN_OK; } static size_t sqlite_mbs_get(struct cla_config *const config) { // UNUSED (void)config; return SIZE_MAX; } static struct cla_tx_queue sqlite_get_tx_queue( struct cla_config *config, const char *eid, const char *cla_addr) { // UNUSED (void)eid; (void)cla_addr; struct sqlite_link *const link = ((struct sqlite_config *)config)->link; // No active link! if (!link) return (struct cla_tx_queue){ NULL, NULL }; hal_semaphore_take_blocking(link->base.tx_queue_sem); // Freed while trying to obtain it if (!link->base.tx_queue_handle) return (struct cla_tx_queue){ NULL, NULL }; return (struct cla_tx_queue){ .tx_queue_handle = link->base.tx_queue_handle, .tx_queue_sem = link->base.tx_queue_sem, }; } static enum ud3tn_result sqlite_start_scheduled_contact( struct cla_config *config, const char *eid, const char *cla_addr) { // UNUSED (void)eid; (void)cla_addr; (void)config; return UD3TN_OK; } static enum ud3tn_result sqlite_end_scheduled_contact( struct cla_config *config, const char *eid, const char *cla_addr) { // UNUSED (void)eid; (void)cla_addr; (void)config; return UD3TN_OK; } static void sqlite_begin_packet( struct cla_link *link, const struct bundle *const bundle, size_t length, char *cla_addr) { // UNUSED (void)cla_addr; LOGF_DEBUG("SQLiteCLA: Storing bundle %p with destination %s", bundle, bundle->destination); struct sqlite_link *sqlite_link = (struct sqlite_link *)link; sqlite3 *db = sqlite_link->db; sqlite3_stmt *stmt; if (sqlite3_prepare_v2(db, SQL_INSERT_INTO, -1, &stmt, 0) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to prepare INSERT statement: %s", sqlite3_errmsg(db)); return; } if (sqlite3_bind_text(stmt, 1, bundle->source, strlen(bundle->source), NULL) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to bind source column: %s", sqlite3_errmsg(db)); return; } if (sqlite3_bind_text( stmt, 2, bundle->destination, strlen(bundle->destination), NULL ) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to bind destination column: %s", sqlite3_errmsg(db)); return; } if (sqlite3_bind_int64(stmt, 3, bundle->creation_timestamp_ms) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to bind destination column: %s", sqlite3_errmsg(db)); return; } if (sqlite3_bind_zeroblob(stmt, 4, length) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to bind bundle column: %s", sqlite3_errmsg(db)); return; } if (sqlite3_step(stmt) != SQLITE_DONE) { LOGF_ERROR("SQLiteCLA: Failed to execute INSERT statement: %s", sqlite3_errmsg(db)); return; }; if (sqlite3_finalize(stmt) != SQLITE_OK) { LOGF_ERROR( "SQLiteCLA: Failed to finalize INSERT statement: %s", sqlite3_errmsg(db) ); return; } /* Open the bundle blob to write the data incrementally when the sqlite_send_packet_data * function is called. */ sqlite_bundle_blob_open(db, sqlite3_last_insert_rowid(db), &sqlite_link->send_bundle_blob); } static void sqlite_send_packet_data( struct cla_link *link, const void *data, const size_t length) { struct sqlite_link *sqlite_link = (struct sqlite_link *)link; sqlite3 *db = sqlite_link->db; struct sqlite_bundle_blob *send_bundle_blob = &sqlite_link->send_bundle_blob; if (sqlite3_blob_write( send_bundle_blob->blob, data, length, send_bundle_blob->offset ) != SQLITE_OK) LOGF_ERROR("SQLiteCLA: Failed to write BLOB: %s", sqlite3_errmsg(db)); send_bundle_blob->offset += length; } static void sqlite_end_packet(struct cla_link *link) { struct sqlite_link *sqlite_link = (struct sqlite_link *)link; sqlite_bundle_blob_close(sqlite_link->db, &sqlite_link->send_bundle_blob); } static void sqlite_reset_parsers(struct cla_link *const link) { struct sqlite_link *sqlite_link = (struct sqlite_link *)link; if (sqlite_link->read_bundle_blob.blob) sqlite_bundle_blob_close(sqlite_link->db, &sqlite_link->read_bundle_blob); rx_task_reset_parsers(&link->rx_task_data); link->rx_task_data.cur_parser = (struct parser *)&link->rx_task_data.bundle7_parser.basedata; } static size_t sqlite_forward_to_specific_parser( struct cla_link *const link, const uint8_t *const buffer, const size_t length) { struct rx_task_data *const rx_data = &link->rx_task_data; size_t result = 0; switch (rx_data->payload_type) { case PAYLOAD_UNKNOWN: result = select_bundle_parser_version(rx_data, buffer, length); if (result == 0) sqlite_reset_parsers(link); break; case PAYLOAD_BUNDLE6: rx_data->cur_parser = rx_data->bundle6_parser.basedata; result = bundle6_parser_read( &rx_data->bundle6_parser, buffer, length ); break; case PAYLOAD_BUNDLE7: rx_data->cur_parser = rx_data->bundle7_parser.basedata; result = bundle7_parser_read( &rx_data->bundle7_parser, buffer, length ); break; default: sqlite_reset_parsers(link); return 0; } return result; } static enum ud3tn_result sqlite_read( struct cla_link *link, uint8_t *buffer, size_t length, size_t *bytes_read) { struct sqlite_link *sqlite_link = (struct sqlite_link *)link; sqlite3 *db = sqlite_link->db; struct sqlite_bundle_blob *read_bundle_blob = &sqlite_link->read_bundle_blob; QueueIdentifier_t agent_queue = sqlite_link->agent_queue; // Check queue for new bundles if no one is in progress. if (!read_bundle_blob->blob) { struct cla_sqlite_command cmd; if (hal_queue_receive(agent_queue, &cmd, -1) == UD3TN_FAIL) { LOG_ERROR("SQLiteCLA: Failed to receive bundle from agent_queue"); return UD3TN_FAIL; } switch (cmd.type) { case SQLITE_COMMAND_SELECT: LOGF_DEBUG( "SQLiteCLA: Received SQLITE_COMMAND_SELECT (row_id = %u)", cmd.row_id ); if (sqlite_bundle_blob_open( db, cmd.row_id, read_bundle_blob ) == UD3TN_FAIL) return UD3TN_FAIL; break; case SQLITE_COMMAND_FINALIZE: LOG_DEBUG("SQLiteCLA: Received SQLITE_COMMAND_FINALIZE: Terminate"); return UD3TN_OK; case SQLITE_COMMAND_UNDEFINED: LOG_DEBUG("SQLiteCLA: Received SQLITE_COMMAND_UNDEFINED"); return UD3TN_FAIL; } } // Recalucate length, if the remaining bundle is smaller than the buffer. if (read_bundle_blob->offset + (sqlite3_int64)length > read_bundle_blob->size) length = read_bundle_blob->size - read_bundle_blob->offset; // Read if (sqlite3_blob_read( read_bundle_blob->blob, buffer, length, read_bundle_blob->offset ) == SQLITE_ERROR) { LOGF_ERROR( "SQLiteCLA: Failed to read Bundle BLOB: %s", sqlite3_errmsg(db) ); sqlite3_blob_close(read_bundle_blob->blob); return UD3TN_FAIL; } // Update the offset read_bundle_blob->offset += length; // The bundle is completely read, reset the buffers to parse the next one. if (read_bundle_blob->offset >= read_bundle_blob->size) sqlite_bundle_blob_close(db, read_bundle_blob); if (bytes_read) *bytes_read = length; return UD3TN_OK; } const struct cla_vtable sqlite_vtable = { .cla_name_get = sqlite_name_get, .cla_launch = sqlite_launch, .cla_terminate = sqlite_terminate, .cla_mbs_get = sqlite_mbs_get, .cla_get_tx_queue = sqlite_get_tx_queue, .cla_start_scheduled_contact = sqlite_start_scheduled_contact, .cla_end_scheduled_contact = sqlite_end_scheduled_contact, .cla_begin_packet = sqlite_begin_packet, .cla_end_packet = sqlite_end_packet, .cla_send_packet_data = sqlite_send_packet_data, .cla_rx_task_reset_parsers = sqlite_reset_parsers, .cla_rx_task_forward_to_specific_parser = sqlite_forward_to_specific_parser, .cla_read = sqlite_read, .cla_disconnect_handler = sqlite_disconnect_handler, }; static enum ud3tn_result sqlite_link_init( struct sqlite_link *link, struct cla_config *base, sqlite3 *db, QueueIdentifier_t agent_queue) { link->db = db; link->agent_queue = agent_queue; sqlite_bundle_blob_reset(&link->read_bundle_blob); sqlite_bundle_blob_reset(&link->send_bundle_blob); if (cla_link_init( &link->base, base, NULL, true, true ) != UD3TN_OK) { LOG_ERROR("SQLiteCLA: Failed to initialize link"); return UD3TN_FAIL; } return UD3TN_OK; } static enum ud3tn_result sqlite_init( struct sqlite_config *config, const char *filename, const struct bundle_agent_interface *bundle_agent_interface) { // Initialize the CLA base configuration if (cla_config_init(&config->base, bundle_agent_interface) != UD3TN_OK) { LOG_ERROR("SQLiteCLA: Failed to initialize CLA base config"); return UD3TN_FAIL; } config->base.vtable = &sqlite_vtable; // Create a message queue to connect the sqlite agent with the CLA QueueIdentifier_t agent_queue = hal_queue_create( CLA_SQLITE_AGENT_QUEUE_LENGTH, sizeof(struct cla_sqlite_command) ); if (!agent_queue) { LOG_ERROR("SQLiteCLA: Failed to create agent queue"); return UD3TN_FAIL; } // Open the database and create the tables sqlite3 *db; if (sqlite3_config(SQLITE_CONFIG_SERIALIZED) != SQLITE_OK) { LOG_ERROR( "SQLiteCLA: Failed to set SQLITE_CONFIG_SERIALIZED (SQLite needs to be compiled with the SQLITE_THREADSAFE=1 compile-time option)" ); goto failed_queue_delete; } if (sqlite3_open(filename, &db) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to open database file: %s", sqlite3_errmsg(db)); goto failed_queue_delete; } if (sqlite3_exec(db, SQL_CREATE_TABLE, 0, 0, NULL) != SQLITE_OK) { LOGF_ERROR("SQLiteCLA: Failed to create table: %s", sqlite3_errmsg(db)); goto failed_db_close; } // Initialize the link configuration config->link = malloc(sizeof(struct sqlite_link)); if (!config->link) { LOG_ERROR("SQLiteCLA: Failed to allocate sqlite_link"); goto failed_db_close; } if (sqlite_link_init(config->link, &config->base, db, agent_queue) != UD3TN_OK) { LOG_ERROR("SQLiteCLA: Failed to initialize sqlite_link"); goto failed_link_free; } // Register the CLA bundle_processor_inform( bundle_agent_interface->bundle_signaling_queue, (struct bundle_processor_signal) { .type = BP_SIGNAL_NEW_LINK_ESTABLISHED, .peer_cla_addr = cla_get_cla_addr_from_link( &config->link->base ), } ); // Initialize the SQLite agent if (sqlite_agent_setup(bundle_agent_interface, db, agent_queue) != 0) { LOG_ERROR("SQLiteCLA: Failed to initialize SQLiteAgent"); goto failed_link_free; } return UD3TN_OK; failed_link_free: free(config->link); failed_db_close: sqlite3_close(db); failed_queue_delete: hal_queue_delete(agent_queue); return UD3TN_FAIL; } struct cla_config *sqlite_create( const char *const options[], const size_t option_count, const struct bundle_agent_interface *bundle_agent_interface) { LOGF_DEBUG("SQLiteCLA: Using library version %s.", sqlite3_libversion()); if (option_count != 1) { LOG_ERROR("SQLiteCLA: Option has to be a file path."); return NULL; } struct sqlite_config *config = malloc(sizeof(struct sqlite_config)); if (!config) { LOG_ERROR("SQLiteCLA: Memory allocation failed!"); return NULL; } if (sqlite_init(config, options[0], bundle_agent_interface) != UD3TN_OK) { free(config); LOG_ERROR("SQLiteCLA: Initialization failed!"); return NULL; } return &config->base; }