#include <MariaDbStorage.h>

Collaboration diagram for mqtt::mqttstore::lib::MariaDbStorage:

Classes
struct	ConnectionConfig

Public Member Functions
	MariaDbStorage (const std::string &connectionName, const ConnectionConfig &connectionConfig, std::string rawTable, bool autoCreateRawTable, StoragePlan storagePlan)
void	store (const MqttMessage &message)

Private Member Functions
void	createRawTable ()
void	storeProjections (const MqttMessage &message, const std::optional< nlohmann::json > &payloadJson)

Static Private Member Functions
static bool	isSafeIdentifier (const std::string &identifier)
static std::string	quoteIdentifier (const std::string &identifier)
static std::string	sqlQuote (const std::string &value)
static std::string	sqlValue (const nlohmann::json &value)
static std::optional< nlohmann::json >	parsePayload (const std::string &payload)
static bool	hasBinaryContent (const std::string &payload)
static std::string	buildRawInsertSql (const std::string &rawTable, const MqttMessage &message, const std::optional< nlohmann::json > &payloadJson)
static std::string	buildProjectionInsertSql (const StoragePlan::Projection &projection, const MqttMessage &message, const nlohmann::json &payloadJson)
static std::string	jsonValueForColumn (const StoragePlan::ColumnMapping &mapping, const MqttMessage &message, const nlohmann::json &payloadJson)
static void	execLogFailure (const std::string &connectionName, const std::string &operation, const std::string &errorString, unsigned int errorNumber)

Private Attributes
std::string	connectionName
database::mariadb::MariaDBClient	mariaDB
std::string	rawTable
StoragePlan	storagePlan

Detailed Description

Definition at line 34 of file MariaDbStorage.h.

Constructor & Destructor Documentation

◆ MariaDbStorage()

mqtt::mqttstore::lib::MariaDbStorage::MariaDbStorage	(	const std::string &	connectionName,
		const ConnectionConfig &	connectionConfig,
		std::string	rawTable,
		bool	autoCreateRawTable,
		StoragePlan	storagePlan )

Definition at line 55 of file MariaDbStorage.cpp.

        : connectionName(connectionName)
        , mariaDB(
              {
                  .connectionName = connectionName,
                  .hostname = connectionConfig.host,
                  .username = connectionConfig.username,
                  .password = connectionConfig.password,
                  .database = connectionConfig.database,
                  .port = connectionConfig.port,
                  .socket = connectionConfig.socket,
                  .flags = connectionConfig.flags,
              },
              [connectionName = this->connectionName](const database::mariadb::MariaDBState& state) {
                  if (state.connected) {
                      VLOG(0) << connectionName << " MariaDB: connected";
                  } else if (state.error != 0) {
                      VLOG(0) << connectionName << " MariaDB: " << state.errorMessage << " [" << state.error << "]";
                  } else {
                      VLOG(0) << connectionName << " MariaDB: lost connection";
                  }
              })
        , rawTable(std::move(rawTable))
        , storagePlan(std::move(storagePlan)) {
        if (!isSafeIdentifier(this->rawTable)) {
            throw std::runtime_error("Unsafe raw table name: " + this->rawTable);
        }
 
        if (autoCreateRawTable) {
            createRawTable();
        }
    }

References connectionName, and MariaDbStorage().

Referenced by MariaDbStorage().

Here is the call graph for this function:

Here is the caller graph for this function:

Member Function Documentation

◆ buildProjectionInsertSql()

std::string mqtt::mqttstore::lib::MariaDbStorage::buildProjectionInsertSql	(	const StoragePlan::Projection &	projection,
		const MqttMessage &	message,
		const nlohmann::json &	payloadJson )

staticnodiscardprivate

Definition at line 209 of file MariaDbStorage.cpp.

                                                                                          {
        std::string columnList;
        std::string valueList;
 
        for (const StoragePlan::ColumnMapping& mapping : projection.columns) {
            const std::string value = jsonValueForColumn(mapping, message, payloadJson);
            if (value.empty() && !mapping.required) {
                continue;
            }
 
            columnList += (columnList.empty() ? "" : ", ") + quoteIdentifier(mapping.column);
            valueList += (valueList.empty() ? "" : ", ") + (value.empty() ? "NULL" : value);
        }
 
        if (columnList.empty()) {
            throw std::runtime_error("Projection '" + projection.name + "' produced no columns");
        }
 
        return "INSERT INTO " + quoteIdentifier(projection.table) + "(" + columnList + ") VALUES (" + valueList + ")";
    }

References mqtt::mqttstore::lib::StoragePlan::Projection::name, quoteIdentifier(), and mqtt::mqttstore::lib::StoragePlan::Projection::table.

Here is the call graph for this function:

◆ buildRawInsertSql()

std::string mqtt::mqttstore::lib::MariaDbStorage::buildRawInsertSql	(	const std::string &	rawTable,
		const MqttMessage &	message,
		const std::optional< nlohmann::json > &	payloadJson )

staticnodiscardprivate

Definition at line 192 of file MariaDbStorage.cpp.

                                                                                                {
        const bool binaryPayload = !payloadJson.has_value() && hasBinaryContent(message.payload);
        const std::string payloadFormat = payloadJson.has_value() ? "json" : (binaryPayload ? "binary" : "text");
        const std::string payloadTextValue = binaryPayload ? "NULL" : sqlQuote(message.payload);
        const std::string payloadJsonValue = payloadJson.has_value() ? sqlQuote(payloadJson->dump()) : "NULL";
 
        return "INSERT INTO " + quoteIdentifier(rawTable) +
               "(`received_at`, `source_instance`, `topic`, `qos`, `retain_flag`, `dup_flag`, `packet_identifier`, `payload`, "
               "`payload_text`, `payload_json`, `payload_format`) VALUES (CURRENT_TIMESTAMP(6), " +
               sqlQuote(message.connectionName) + ", " + sqlQuote(message.topic) + ", " +
               std::to_string(static_cast<unsigned int>(message.qoS)) + ", " + (message.retain ? "TRUE" : "FALSE") + ", " +
               (message.dup ? "TRUE" : "FALSE") + ", " + std::to_string(message.packetIdentifier) + ", " + sqlQuote(message.payload) +
               ", " + payloadTextValue + ", " + payloadJsonValue + ", " + sqlQuote(payloadFormat) + ")";
    }

References mqtt::mqttstore::lib::MqttMessage::connectionName, mqtt::mqttstore::lib::MqttMessage::dup, hasBinaryContent(), mqtt::mqttstore::lib::MqttMessage::packetIdentifier, mqtt::mqttstore::lib::MqttMessage::payload, mqtt::mqttstore::lib::MqttMessage::qoS, quoteIdentifier(), mqtt::mqttstore::lib::MqttMessage::retain, sqlQuote(), and mqtt::mqttstore::lib::MqttMessage::topic.

Referenced by store().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ createRawTable()

void mqtt::mqttstore::lib::MariaDbStorage::createRawTable ( )

private

Definition at line 267 of file MariaDbStorage.cpp.

                                        {
        const std::string sql = "CREATE TABLE IF NOT EXISTS " + quoteIdentifier(rawTable) +
                                "("
                                "`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT PRIMARY KEY,"
                                "`received_at` TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),"
                                "`source_instance` VARCHAR(255) NULL,"
                                "`topic` VARCHAR(1024) NOT NULL,"
                                "`qos` TINYINT UNSIGNED NOT NULL,"
                                "`retain_flag` BOOLEAN NOT NULL,"
                                "`dup_flag` BOOLEAN NOT NULL,"
                                "`packet_identifier` INT UNSIGNED NULL,"
                                "`payload` LONGBLOB NOT NULL,"
                                "`payload_text` LONGTEXT NULL,"
                                "`payload_json` JSON NULL,"
                                "`payload_format` ENUM('json', 'text', 'binary') NOT NULL,"
                                "INDEX `idx_received_at` (`received_at`),"
                                "INDEX `idx_topic` (`topic`(255))"
                                ")";
 
        mariaDB.exec(
            sql,
            [connectionName = this->connectionName, rawTable = this->rawTable]() -> void {
                VLOG(0) << connectionName << " MariaDB: ensured raw MQTT table '" << rawTable << "'";
            },
            [connectionName = this->connectionName](const std::string& errorString, unsigned int errorNumber) -> void {
                execLogFailure(connectionName, "raw MQTT table creation", errorString, errorNumber);
            });
    }

References quoteIdentifier(), and rawTable.

Here is the call graph for this function:

◆ execLogFailure()

void mqtt::mqttstore::lib::MariaDbStorage::execLogFailure	(	const std::string &	connectionName,
		const std::string &	operation,
		const std::string &	errorString,
		unsigned int	errorNumber )

staticprivate

Definition at line 260 of file MariaDbStorage.cpp.

                                                                  {
        VLOG(0) << connectionName << " MariaDB " << operation << " failed: " << errorString << " : " << errorNumber;
    }

◆ hasBinaryContent()

bool mqtt::mqttstore::lib::MariaDbStorage::hasBinaryContent ( const std::string & payload )

staticnodiscardprivate

Definition at line 186 of file MariaDbStorage.cpp.

                                                                  {
        return std::any_of(payload.begin(), payload.end(), [](unsigned char character) {
            return character == '\0' || (character < 0x09) || (character > 0x0D && character < 0x20);
        });
    }

Referenced by buildRawInsertSql().

Here is the caller graph for this function:

◆ isSafeIdentifier()

bool mqtt::mqttstore::lib::MariaDbStorage::isSafeIdentifier ( const std::string & identifier )

staticnodiscardprivate

Definition at line 108 of file MariaDbStorage.cpp.

                                                                     {
        return !identifier.empty() && std::all_of(identifier.begin(), identifier.end(), [](unsigned char character) {
            return std::isalnum(character) != 0 || character == '_';
        });
    }

Referenced by quoteIdentifier().

Here is the caller graph for this function:

◆ jsonValueForColumn()

std::string mqtt::mqttstore::lib::MariaDbStorage::jsonValueForColumn	(	const StoragePlan::ColumnMapping &	mapping,
		const MqttMessage &	message,
		const nlohmann::json &	payloadJson )

staticnodiscardprivate

Definition at line 232 of file MariaDbStorage.cpp.

                                                                                    {
        if (mapping.literal.has_value()) {
            return sqlQuote(*mapping.literal);
        }
 
        if (mapping.topicLevel.has_value()) {
            const std::vector<std::string> topicLevels = splitTopic(message.topic);
            if (*mapping.topicLevel >= topicLevels.size()) {
                return {};
            }
 
            return sqlQuote(topicLevels[*mapping.topicLevel]);
        }
 
        if (mapping.jsonPointer.empty()) {
            return {};
        }
 
        const nlohmann::json::json_pointer pointer(mapping.jsonPointer);
        if (!payloadJson.contains(pointer)) {
            return {};
        }
 
        return sqlValue(payloadJson.at(pointer));
    }

References mqtt::mqttstore::lib::StoragePlan::ColumnMapping::jsonPointer, mqtt::mqttstore::lib::StoragePlan::ColumnMapping::literal, sqlQuote(), sqlValue(), and mqtt::mqttstore::lib::StoragePlan::ColumnMapping::topicLevel.

Here is the call graph for this function:

◆ parsePayload()

std::optional< nlohmann::json > mqtt::mqttstore::lib::MariaDbStorage::parsePayload ( const std::string & payload )

staticnodiscardprivate

Definition at line 178 of file MariaDbStorage.cpp.

                                                                                   {
        try {
            return nlohmann::json::parse(payload);
        } catch (const nlohmann::json::parse_error&) {
            return std::nullopt;
        }
    }

◆ quoteIdentifier()

std::string mqtt::mqttstore::lib::MariaDbStorage::quoteIdentifier ( const std::string & identifier )

staticnodiscardprivate

Definition at line 114 of file MariaDbStorage.cpp.

                                                                         {
        if (!isSafeIdentifier(identifier)) {
            throw std::runtime_error("Unsafe SQL identifier: " + identifier);
        }
 
        return "`" + identifier + "`";
    }

References isSafeIdentifier().

Referenced by buildProjectionInsertSql(), buildRawInsertSql(), and createRawTable().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ sqlQuote()

std::string mqtt::mqttstore::lib::MariaDbStorage::sqlQuote ( const std::string & value )

staticnodiscardprivate

Definition at line 122 of file MariaDbStorage.cpp.

                                                             {
        std::string quoted;
        quoted.reserve(value.size() + 2);
        quoted.push_back('\'');
 
        for (const char character : value) {
            switch (character) {
                case '\0':
                    quoted += "\\0";
                    break;
                case '\n':
                    quoted += "\\n";
                    break;
                case '\r':
                    quoted += "\\r";
                    break;
                case '\\':
                    quoted += "\\\\";
                    break;
                case '\'':
                    quoted += "\\'";
                    break;
                case '"':
                    quoted += "\\\"";
                    break;
                case '\x1a':
                    quoted += "\\Z";
                    break;
                default:
                    quoted.push_back(character);
                    break;
            }
        }
 
        quoted.push_back('\'');
 
        return quoted;
    }

Referenced by buildRawInsertSql(), jsonValueForColumn(), and sqlValue().

Here is the caller graph for this function:

◆ sqlValue()

std::string mqtt::mqttstore::lib::MariaDbStorage::sqlValue ( const nlohmann::json & value )

staticnodiscardprivate

Definition at line 161 of file MariaDbStorage.cpp.

                                                                {
        if (value.is_null()) {
            return "NULL";
        }
        if (value.is_boolean()) {
            return value.get<bool>() ? "TRUE" : "FALSE";
        }
        if (value.is_number()) {
            return value.dump();
        }
        if (value.is_string()) {
            return sqlQuote(value.get<std::string>());
        }
 
        return sqlQuote(value.dump());
    }

References sqlQuote().

Referenced by jsonValueForColumn().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ store()

void mqtt::mqttstore::lib::MariaDbStorage::store ( const MqttMessage & message )

Definition at line 92 of file MariaDbStorage.cpp.

                                                         {
        const std::optional<nlohmann::json> payloadJson = parsePayload(message.payload);
        const std::string rawInsertSql = buildRawInsertSql(rawTable, message, payloadJson);
 
        mariaDB.exec(
            rawInsertSql,
            [connectionName = this->connectionName, topic = message.topic]() -> void {
                VLOG(1) << connectionName << " MariaDB: stored raw MQTT message for topic '" << topic << "'";
            },
            [connectionName = this->connectionName](const std::string& errorString, unsigned int errorNumber) -> void {
                execLogFailure(connectionName, "raw MQTT message insert", errorString, errorNumber);
            });
 
        storeProjections(message, payloadJson);
    }

References buildRawInsertSql(), rawTable, and storeProjections().

Here is the call graph for this function:

◆ storeProjections()

void mqtt::mqttstore::lib::MariaDbStorage::storeProjections	(	const MqttMessage &	message,
		const std::optional< nlohmann::json > &	payloadJson )

private

Definition at line 296 of file MariaDbStorage.cpp.

                                                                                                                  {
        if (!payloadJson.has_value()) {
            return;
        }
 
        for (const StoragePlan::Projection* projection : storagePlan.match(message.topic)) {
            try {
                const std::string sql = buildProjectionInsertSql(*projection, message, *payloadJson);
                mariaDB.exec(
                    sql,
                    [connectionName = this->connectionName, projectionName = projection->name]() -> void {
                        VLOG(1) << connectionName << " MariaDB: projection insert completed for '" << projectionName << "'";
                    },
                    [connectionName = this->connectionName, projectionName = projection->name](const std::string& errorString,
                                                                                               unsigned int errorNumber) -> void {
                        execLogFailure(connectionName, "projection '" + projectionName + "' insert", errorString, errorNumber);
                    });
            } catch (const std::exception& error) {
                VLOG(0) << connectionName << " MariaDB projection '" << projection->name << "' skipped: " << error.what();
            }
        }
    }

Referenced by store().

Here is the caller graph for this function:

Member Data Documentation

◆ connectionName

std::string mqtt::mqttstore::lib::MariaDbStorage::connectionName

private

Definition at line 74 of file MariaDbStorage.h.

Referenced by MariaDbStorage().

◆ mariaDB

database::mariadb::MariaDBClient mqtt::mqttstore::lib::MariaDbStorage::mariaDB

private

Definition at line 75 of file MariaDbStorage.h.

◆ rawTable

std::string mqtt::mqttstore::lib::MariaDbStorage::rawTable

private

Definition at line 76 of file MariaDbStorage.h.

Referenced by createRawTable(), and store().

◆ storagePlan

StoragePlan mqtt::mqttstore::lib::MariaDbStorage::storagePlan

private

Definition at line 77 of file MariaDbStorage.h.

The documentation for this class was generated from the following files:

mqttstore/lib/MariaDbStorage.h
mqttstore/lib/MariaDbStorage.cpp

MQTTSuite

Classes

Public Member Functions

Private Member Functions

Static Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

◆ MariaDbStorage()

Member Function Documentation

◆ buildProjectionInsertSql()

◆ buildRawInsertSql()

◆ createRawTable()

◆ execLogFailure()

◆ hasBinaryContent()

◆ isSafeIdentifier()

◆ jsonValueForColumn()

◆ parsePayload()

◆ quoteIdentifier()

◆ sqlQuote()

◆ sqlValue()

◆ store()

◆ storeProjections()

Member Data Documentation

◆ connectionName

◆ mariaDB

◆ rawTable

◆ storagePlan