|
- /*
- * This file is part of GNUnet
- * Copyright (C) 2016 GNUnet e.V.
- *
- * GNUnet is free software: you can redistribute it and/or modify it
- * under the terms of the GNU Affero General Public License as published
- * by the Free Software Foundation, either version 3 of the License,
- * or (at your option) any later version.
- *
- * GNUnet is distributed in the hope that it will be useful, but
- * WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program. If not, see <http://www.gnu.org/licenses/>.
-
- SPDX-License-Identifier: AGPL3.0-or-later
- */
-
- /**
- * @file psycstore/plugin_psycstore_postgres.c
- * @brief PostgresQL-based psycstore backend
- * @author Daniel Golle
- * @author Gabor X Toth
- * @author Christian Grothoff
- * @author Christophe Genevey
- * @author Jeffrey Burdges
- */
-
- #include <gnunet/platform.h>
- #include "gnunet_psycstore_plugin.h"
- #include "gnunet_psycstore_service.h"
- #include "gnunet_multicast_service.h"
- #include <gnunet/gnunet_crypto_lib.h>
- #include "gnunet_psyc_util_lib.h"
- #include "psycstore.h"
- #include <gnunet/gnunet_pq_lib.h>
-
- /**
- * After how many ms "busy" should a DB operation fail for good? A
- * low value makes sure that we are more responsive to requests
- * (especially PUTs). A high value guarantees a higher success rate
- * (SELECTs in iterate can take several seconds despite LIMIT=1).
- *
- * The default value of 1s should ensure that users do not experience
- * huge latencies while at the same time allowing operations to
- * succeed with reasonable probability.
- */
- #define BUSY_TIMEOUT_MS 1000
-
- #define DEBUG_PSYCSTORE GNUNET_EXTRA_LOGGING
-
- #define LOG(kind,...) GNUNET_log_from (kind, "psycstore-postgres", __VA_ARGS__)
-
- enum Transactions {
- TRANSACTION_NONE = 0,
- TRANSACTION_STATE_MODIFY,
- TRANSACTION_STATE_SYNC,
- };
-
- /**
- * Context for all functions in this plugin.
- */
- struct Plugin
- {
-
- const struct GNUNET_CONFIGURATION_Handle *cfg;
-
- /**
- * Native Postgres database handle.
- */
- PGconn *dbh;
-
- enum Transactions transaction;
-
- void *cls;
- };
-
-
- /**
- * Initialize the database connections and associated
- * data structures (create tables and indices
- * as needed as well).
- *
- * @param plugin the plugin context (state for this module)
- * @return #GNUNET_OK on success
- */
- static int
- database_setup (struct Plugin *plugin)
- {
- struct GNUNET_PQ_ExecuteStatement es[] = {
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS channels (\n"
- " id SERIAL,\n"
- " pub_key BYTEA NOT NULL CHECK (LENGTH(pub_key)=32),\n"
- " max_state_message_id BIGINT,\n"
- " state_hash_message_id BIGINT,\n"
- " PRIMARY KEY(id)\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_make_execute ("CREATE UNIQUE INDEX IF NOT EXISTS channel_pub_key_idx \n"
- " ON channels (pub_key)"),
- GNUNET_PQ_make_execute ("CREATE OR REPLACE FUNCTION get_chan_id(BYTEA) RETURNS INTEGER AS \n"
- " 'SELECT id FROM channels WHERE pub_key=$1;' LANGUAGE SQL STABLE \n"
- "RETURNS NULL ON NULL INPUT"),
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS slaves (\n"
- " id SERIAL,\n"
- " pub_key BYTEA NOT NULL CHECK (LENGTH(pub_key)=32),\n"
- " PRIMARY KEY(id)\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_make_execute ("CREATE UNIQUE INDEX IF NOT EXISTS slaves_pub_key_idx \n"
- " ON slaves (pub_key)"),
- GNUNET_PQ_make_execute ("CREATE OR REPLACE FUNCTION get_slave_id(BYTEA) RETURNS INTEGER AS \n"
- " 'SELECT id FROM slaves WHERE pub_key=$1;' LANGUAGE SQL STABLE \n"
- "RETURNS NULL ON NULL INPUT"),
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS membership (\n"
- " channel_id BIGINT NOT NULL REFERENCES channels(id),\n"
- " slave_id BIGINT NOT NULL REFERENCES slaves(id),\n"
- " did_join INT NOT NULL,\n"
- " announced_at BIGINT NOT NULL,\n"
- " effective_since BIGINT NOT NULL,\n"
- " group_generation BIGINT NOT NULL\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_make_execute ("CREATE INDEX IF NOT EXISTS idx_membership_channel_id_slave_id "
- "ON membership (channel_id, slave_id)"),
- /** @todo messages table: add method_name column */
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS messages (\n"
- " channel_id BIGINT NOT NULL REFERENCES channels(id),\n"
- " hop_counter INT NOT NULL,\n"
- " signature BYTEA CHECK (LENGTH(signature)=64),\n"
- " purpose BYTEA CHECK (LENGTH(purpose)=8),\n"
- " fragment_id BIGINT NOT NULL,\n"
- " fragment_offset BIGINT NOT NULL,\n"
- " message_id BIGINT NOT NULL,\n"
- " group_generation BIGINT NOT NULL,\n"
- " multicast_flags INT NOT NULL,\n"
- " psycstore_flags INT NOT NULL,\n"
- " data BYTEA,\n"
- " PRIMARY KEY (channel_id, fragment_id),\n"
- " UNIQUE (channel_id, message_id, fragment_offset)\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS state (\n"
- " channel_id BIGINT NOT NULL REFERENCES channels(id),\n"
- " name TEXT NOT NULL,\n"
- " value_current BYTEA,\n"
- " value_signed BYTEA,\n"
- " PRIMARY KEY (channel_id, name)\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_make_execute ("CREATE TABLE IF NOT EXISTS state_sync (\n"
- " channel_id BIGINT NOT NULL REFERENCES channels(id),\n"
- " name TEXT NOT NULL,\n"
- " value BYTEA,\n"
- " PRIMARY KEY (channel_id, name)\n"
- ")"
- "WITH OIDS"),
- GNUNET_PQ_EXECUTE_STATEMENT_END
- };
-
- /* Open database and precompile statements */
- plugin->dbh = GNUNET_PQ_connect_with_cfg (plugin->cfg,
- "psycstore-postgres");
- if (NULL == plugin->dbh)
- return GNUNET_SYSERR;
- if (GNUNET_OK !=
- GNUNET_PQ_exec_statements (plugin->dbh,
- es))
- {
- PQfinish (plugin->dbh);
- plugin->dbh = NULL;
- return GNUNET_SYSERR;
- }
-
- /* Prepare statements */
- {
- struct GNUNET_PQ_PreparedStatement ps[] = {
- GNUNET_PQ_make_prepare ("transaction_begin",
- "BEGIN", 0),
- GNUNET_PQ_make_prepare ("transaction_commit",
- "COMMIT", 0),
- GNUNET_PQ_make_prepare ("transaction_rollback",
- "ROLLBACK", 0),
- GNUNET_PQ_make_prepare ("insert_channel_key",
- "INSERT INTO channels (pub_key) VALUES ($1)"
- " ON CONFLICT DO NOTHING", 1),
- GNUNET_PQ_make_prepare ("insert_slave_key",
- "INSERT INTO slaves (pub_key) VALUES ($1)"
- " ON CONFLICT DO NOTHING", 1),
- GNUNET_PQ_make_prepare ("insert_membership",
- "INSERT INTO membership\n"
- " (channel_id, slave_id, did_join, announced_at,\n"
- " effective_since, group_generation)\n"
- "VALUES (get_chan_id($1),\n"
- " get_slave_id($2),\n"
- " $3, $4, $5, $6)", 6),
- GNUNET_PQ_make_prepare ("select_membership",
- "SELECT did_join FROM membership\n"
- "WHERE channel_id = get_chan_id($1)\n"
- " AND slave_id = get_slave_id($2)\n"
- " AND effective_since <= $3 AND did_join = 1\n"
- "ORDER BY announced_at DESC LIMIT 1", 3),
- GNUNET_PQ_make_prepare ("insert_fragment",
- "INSERT INTO messages\n"
- " (channel_id, hop_counter, signature, purpose,\n"
- " fragment_id, fragment_offset, message_id,\n"
- " group_generation, multicast_flags, psycstore_flags, data)\n"
- "VALUES (get_chan_id($1),\n"
- " $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)"
- "ON CONFLICT DO NOTHING", 11),
- GNUNET_PQ_make_prepare ("update_message_flags",
- "UPDATE messages\n"
- "SET psycstore_flags = psycstore_flags | $1\n"
- "WHERE channel_id = get_chan_id($2) \n"
- " AND message_id = $3 AND fragment_offset = 0", 3),
- GNUNET_PQ_make_prepare ("select_fragments",
- "SELECT hop_counter, signature, purpose, fragment_id,\n"
- " fragment_offset, message_id, group_generation,\n"
- " multicast_flags, psycstore_flags, data\n"
- "FROM messages\n"
- "WHERE channel_id = get_chan_id($1) \n"
- " AND $2 <= fragment_id AND fragment_id <= $3", 3),
- /** @todo select_messages: add method_prefix filter */
- GNUNET_PQ_make_prepare ("select_messages",
- "SELECT hop_counter, signature, purpose, fragment_id,\n"
- " fragment_offset, message_id, group_generation,\n"
- " multicast_flags, psycstore_flags, data\n"
- "FROM messages\n"
- "WHERE channel_id = get_chan_id($1) \n"
- " AND $2 <= message_id AND message_id <= $3\n"
- "LIMIT $4;", 4),
- /** @todo select_latest_messages: add method_prefix filter */
- GNUNET_PQ_make_prepare ("select_latest_fragments",
- "SELECT rev.hop_counter AS hop_counter,\n"
- " rev.signature AS signature,\n"
- " rev.purpose AS purpose,\n"
- " rev.fragment_id AS fragment_id,\n"
- " rev.fragment_offset AS fragment_offset,\n"
- " rev.message_id AS message_id,\n"
- " rev.group_generation AS group_generation,\n"
- " rev.multicast_flags AS multicast_flags,\n"
- " rev.psycstore_flags AS psycstore_flags,\n"
- " rev.data AS data\n"
- " FROM\n"
- " (SELECT hop_counter, signature, purpose, fragment_id,\n"
- " fragment_offset, message_id, group_generation,\n"
- " multicast_flags, psycstore_flags, data \n"
- " FROM messages\n"
- " WHERE channel_id = get_chan_id($1) \n"
- " ORDER BY fragment_id DESC\n"
- " LIMIT $2) AS rev\n"
- " ORDER BY rev.fragment_id;", 2),
- GNUNET_PQ_make_prepare ("select_latest_messages",
- "SELECT hop_counter, signature, purpose, fragment_id,\n"
- " fragment_offset, message_id, group_generation,\n"
- " multicast_flags, psycstore_flags, data\n"
- "FROM messages\n"
- "WHERE channel_id = get_chan_id($1)\n"
- " AND message_id IN\n"
- " (SELECT message_id\n"
- " FROM messages\n"
- " WHERE channel_id = get_chan_id($2) \n"
- " GROUP BY message_id\n"
- " ORDER BY message_id\n"
- " DESC LIMIT $3)\n"
- "ORDER BY fragment_id", 3),
- GNUNET_PQ_make_prepare ("select_message_fragment",
- "SELECT hop_counter, signature, purpose, fragment_id,\n"
- " fragment_offset, message_id, group_generation,\n"
- " multicast_flags, psycstore_flags, data\n"
- "FROM messages\n"
- "WHERE channel_id = get_chan_id($1) \n"
- " AND message_id = $2 AND fragment_offset = $3", 3),
- GNUNET_PQ_make_prepare ("select_counters_message",
- "SELECT fragment_id, message_id, group_generation\n"
- "FROM messages\n"
- "WHERE channel_id = get_chan_id($1)\n"
- "ORDER BY fragment_id DESC LIMIT 1", 1),
- GNUNET_PQ_make_prepare ("select_counters_state",
- "SELECT max_state_message_id\n"
- "FROM channels\n"
- "WHERE pub_key = $1 AND max_state_message_id IS NOT NULL", 1),
- GNUNET_PQ_make_prepare ("update_max_state_message_id",
- "UPDATE channels\n"
- "SET max_state_message_id = $1\n"
- "WHERE pub_key = $2", 2),
-
- GNUNET_PQ_make_prepare ("update_state_hash_message_id",
- "UPDATE channels\n"
- "SET state_hash_message_id = $1\n"
- "WHERE pub_key = $2", 2),
- GNUNET_PQ_make_prepare ("insert_state_current",
- "INSERT INTO state\n"
- " (channel_id, name, value_current, value_signed)\n"
- "SELECT new.channel_id, new.name,\n"
- " new.value_current, old.value_signed\n"
- "FROM (SELECT get_chan_id($1) AS channel_id,\n"
- " $2::TEXT AS name, $3::BYTEA AS value_current) AS new\n"
- "LEFT JOIN (SELECT channel_id, name, value_signed\n"
- " FROM state) AS old\n"
- "ON new.channel_id = old.channel_id AND new.name = old.name\n"
- "ON CONFLICT (channel_id, name)\n"
- " DO UPDATE SET value_current = EXCLUDED.value_current,\n"
- " value_signed = EXCLUDED.value_signed", 3),
- GNUNET_PQ_make_prepare ("delete_state_empty",
- "DELETE FROM state\n"
- "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = $1)\n"
- " AND (value_current IS NULL OR length(value_current) = 0)\n"
- " AND (value_signed IS NULL OR length(value_signed) = 0)", 1),
- GNUNET_PQ_make_prepare ("update_state_signed",
- "UPDATE state\n"
- "SET value_signed = value_current\n"
- "WHERE channel_id = get_chan_id($1) ", 1),
- GNUNET_PQ_make_prepare ("delete_state",
- "DELETE FROM state\n"
- "WHERE channel_id = get_chan_id($1) ", 1),
- GNUNET_PQ_make_prepare ("insert_state_sync",
- "INSERT INTO state_sync (channel_id, name, value)\n"
- "VALUES (get_chan_id($1), $2, $3)", 3),
- GNUNET_PQ_make_prepare ("insert_state_from_sync",
- "INSERT INTO state\n"
- " (channel_id, name, value_current, value_signed)\n"
- "SELECT channel_id, name, value, value\n"
- "FROM state_sync\n"
- "WHERE channel_id = get_chan_id($1)", 1),
- GNUNET_PQ_make_prepare ("delete_state_sync",
- "DELETE FROM state_sync\n"
- "WHERE channel_id = get_chan_id($1)", 1),
- GNUNET_PQ_make_prepare ("select_state_one",
- "SELECT value_current\n"
- "FROM state\n"
- "WHERE channel_id = get_chan_id($1)\n"
- " AND name = $2", 2),
- GNUNET_PQ_make_prepare ("select_state_prefix",
- "SELECT name, value_current\n"
- "FROM state\n"
- "WHERE channel_id = get_chan_id($1)\n"
- " AND (name = $2 OR substr(name, 1, $3) = $4)", 4),
- GNUNET_PQ_make_prepare ("select_state_signed",
- "SELECT name, value_signed\n"
- "FROM state\n"
- "WHERE channel_id = get_chan_id($1)\n"
- " AND value_signed IS NOT NULL", 1),
- GNUNET_PQ_PREPARED_STATEMENT_END
- };
-
- if (GNUNET_OK !=
- GNUNET_PQ_prepare_statements (plugin->dbh,
- ps))
- {
- PQfinish (plugin->dbh);
- plugin->dbh = NULL;
- return GNUNET_SYSERR;
- }
- }
-
- return GNUNET_OK;
- }
-
-
- /**
- * Shutdown database connection and associate data
- * structures.
- * @param plugin the plugin context (state for this module)
- */
- static void
- database_shutdown (struct Plugin *plugin)
- {
- PQfinish (plugin->dbh);
- plugin->dbh = NULL;
- }
-
-
- /**
- * Execute a prepared statement with a @a channel_key argument.
- *
- * @param plugin Plugin handle.
- * @param stmt Statement to execute.
- * @param channel_key Public key of the channel.
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- exec_channel (struct Plugin *plugin, const char *stmt,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, stmt, params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- /**
- * Begin a transaction.
- */
- static int
- transaction_begin (struct Plugin *plugin, enum Transactions transaction)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "transaction_begin", params))
- return GNUNET_SYSERR;
-
- plugin->transaction = transaction;
- return GNUNET_OK;
- }
-
-
- /**
- * Commit current transaction.
- */
- static int
- transaction_commit (struct Plugin *plugin)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "transaction_commit", params))
- return GNUNET_SYSERR;
-
- plugin->transaction = TRANSACTION_NONE;
- return GNUNET_OK;
- }
-
-
- /**
- * Roll back current transaction.
- */
- static int
- transaction_rollback (struct Plugin *plugin)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "transaction_rollback", params))
- return GNUNET_SYSERR;
-
- plugin->transaction = TRANSACTION_NONE;
- return GNUNET_OK;
- }
-
-
- static int
- channel_key_store (struct Plugin *plugin,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh,
- "insert_channel_key",
- params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- static int
- slave_key_store (struct Plugin *plugin,
- const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (slave_key),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "insert_slave_key", params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- /**
- * Store join/leave events for a PSYC channel in order to be able to answer
- * membership test queries later.
- *
- * @see GNUNET_PSYCSTORE_membership_store()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- postgres_membership_store (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
- int did_join,
- uint64_t announced_at,
- uint64_t effective_since,
- uint64_t group_generation)
- {
- struct Plugin *plugin = cls;
- uint32_t idid_join = (uint32_t) did_join;
-
- GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
-
- if ( (announced_at > INT64_MAX) ||
- (effective_since > INT64_MAX) ||
- (group_generation > INT64_MAX) )
- {
- GNUNET_break (0);
- return GNUNET_SYSERR;
- }
-
- if ( (GNUNET_OK !=
- channel_key_store (plugin, channel_key)) ||
- (GNUNET_OK !=
- slave_key_store (plugin, slave_key)) )
- return GNUNET_SYSERR;
-
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_auto_from_type (slave_key),
- GNUNET_PQ_query_param_uint32 (&idid_join),
- GNUNET_PQ_query_param_uint64 (&announced_at),
- GNUNET_PQ_query_param_uint64 (&effective_since),
- GNUNET_PQ_query_param_uint64 (&group_generation),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh,
- "insert_membership",
- params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
- /**
- * Test if a member was admitted to the channel at the given message ID.
- *
- * @see GNUNET_PSYCSTORE_membership_test()
- *
- * @return #GNUNET_YES if the member was admitted, #GNUNET_NO if not,
- * #GNUNET_SYSERR if there was en error.
- */
- static int
- membership_test (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
- uint64_t message_id)
- {
- struct Plugin *plugin = cls;
-
- uint32_t did_join = 0;
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_auto_from_type (slave_key),
- GNUNET_PQ_query_param_uint64 (&message_id),
- GNUNET_PQ_query_param_end
- };
-
- struct GNUNET_PQ_ResultSpec results_select[] = {
- GNUNET_PQ_result_spec_uint32 ("did_join", &did_join),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_singleton_select (plugin->dbh, "select_membership",
- params_select, results_select))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
- /**
- * Store a message fragment sent to a channel.
- *
- * @see GNUNET_PSYCSTORE_fragment_store()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- fragment_store (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const struct GNUNET_MULTICAST_MessageHeader *msg,
- uint32_t psycstore_flags)
- {
- struct Plugin *plugin = cls;
-
- GNUNET_assert (TRANSACTION_NONE == plugin->transaction);
-
- uint64_t fragment_id = GNUNET_ntohll (msg->fragment_id);
-
- uint64_t fragment_offset = GNUNET_ntohll (msg->fragment_offset);
- uint64_t message_id = GNUNET_ntohll (msg->message_id);
- uint64_t group_generation = GNUNET_ntohll (msg->group_generation);
-
- uint32_t hop_counter = ntohl(msg->hop_counter);
- uint32_t flags = ntohl(msg->flags);
-
- if (fragment_id > INT64_MAX || fragment_offset > INT64_MAX ||
- message_id > INT64_MAX || group_generation > INT64_MAX)
- {
- LOG(GNUNET_ERROR_TYPE_ERROR,
- "Tried to store fragment with a field > INT64_MAX: "
- "%lu, %lu, %lu, %lu\n", fragment_id, fragment_offset,
- message_id, group_generation);
- GNUNET_break (0);
- return GNUNET_SYSERR;
- }
-
- if (GNUNET_OK != channel_key_store (plugin, channel_key))
- return GNUNET_SYSERR;
-
- struct GNUNET_PQ_QueryParam params_insert[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint32 (&hop_counter),
- GNUNET_PQ_query_param_auto_from_type (&msg->signature),
- GNUNET_PQ_query_param_auto_from_type (&msg->purpose),
- GNUNET_PQ_query_param_uint64 (&fragment_id),
- GNUNET_PQ_query_param_uint64 (&fragment_offset),
- GNUNET_PQ_query_param_uint64 (&message_id),
- GNUNET_PQ_query_param_uint64 (&group_generation),
- GNUNET_PQ_query_param_uint32 (&flags),
- GNUNET_PQ_query_param_uint32 (&psycstore_flags),
- GNUNET_PQ_query_param_fixed_size (&msg[1], ntohs (msg->header.size) - sizeof (*msg)),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "insert_fragment", params_insert))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
- /**
- * Set additional flags for a given message.
- *
- * They are OR'd with any existing flags set.
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- message_add_flags (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_id,
- uint32_t psycstore_flags)
- {
- struct Plugin *plugin = cls;
-
- struct GNUNET_PQ_QueryParam params_update[] = {
- GNUNET_PQ_query_param_uint32 (&psycstore_flags),
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&message_id),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, "update_message_flags", params_update))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- /**
- * Closure for #fragment_rows.
- */
- struct FragmentRowsContext {
- GNUNET_PSYCSTORE_FragmentCallback cb;
- void *cb_cls;
-
- uint64_t *returned_fragments;
-
- /* I preserved this but I do not see the point since
- * it cannot stop the loop early and gets overwritten ?? */
- int ret;
- };
-
-
- /**
- * Callback that retrieves the results of a SELECT statement
- * reading form the messages table.
- *
- * Only passed to GNUNET_PQ_eval_prepared_multi_select and
- * has type GNUNET_PQ_PostgresResultHandler.
- *
- * @param cls closure
- * @param result the postgres result
- * @param num_result the number of results in @a result
- */
- void fragment_rows (void *cls,
- PGresult *res,
- unsigned int num_results)
- {
- struct FragmentRowsContext *c = cls;
-
- for (unsigned int i=0;i<num_results;i++)
- {
- uint32_t hop_counter;
- void *signature = NULL;
- void *purpose = NULL;
- size_t signature_size;
- size_t purpose_size;
- uint64_t fragment_id;
- uint64_t fragment_offset;
- uint64_t message_id;
- uint64_t group_generation;
- uint32_t flags;
- void *buf;
- size_t buf_size;
- uint32_t msg_flags;
- struct GNUNET_PQ_ResultSpec results[] = {
- GNUNET_PQ_result_spec_uint32 ("hop_counter", &hop_counter),
- GNUNET_PQ_result_spec_variable_size ("signature", &signature, &signature_size),
- GNUNET_PQ_result_spec_variable_size ("purpose", &purpose, &purpose_size),
- GNUNET_PQ_result_spec_uint64 ("fragment_id", &fragment_id),
- GNUNET_PQ_result_spec_uint64 ("fragment_offset", &fragment_offset),
- GNUNET_PQ_result_spec_uint64 ("message_id", &message_id),
- GNUNET_PQ_result_spec_uint64 ("group_generation", &group_generation),
- GNUNET_PQ_result_spec_uint32 ("multicast_flags", &msg_flags),
- GNUNET_PQ_result_spec_uint32 ("psycstore_flags", &flags),
- GNUNET_PQ_result_spec_variable_size ("data", &buf, &buf_size),
- GNUNET_PQ_result_spec_end
- };
- struct GNUNET_MULTICAST_MessageHeader *mp;
-
- if (GNUNET_YES != GNUNET_PQ_extract_result (res, results, i))
- {
- GNUNET_PQ_cleanup_result(results); /* missing previously, a memory leak?? */
- break; /* nothing more?? */
- }
-
- mp = GNUNET_malloc (sizeof (*mp) + buf_size);
-
- mp->header.size = htons (sizeof (*mp) + buf_size);
- mp->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE);
- mp->hop_counter = htonl (hop_counter);
- GNUNET_memcpy (&mp->signature,
- signature, signature_size);
- GNUNET_memcpy (&mp->purpose,
- purpose, purpose_size);
- mp->fragment_id = GNUNET_htonll (fragment_id);
- mp->fragment_offset = GNUNET_htonll (fragment_offset);
- mp->message_id = GNUNET_htonll (message_id);
- mp->group_generation = GNUNET_htonll (group_generation);
- mp->flags = htonl(msg_flags);
-
- GNUNET_memcpy (&mp[1],
- buf, buf_size);
- GNUNET_PQ_cleanup_result(results);
- c->ret = c->cb (c->cb_cls, mp, (enum GNUNET_PSYCSTORE_MessageFlags) flags);
- if (NULL != c->returned_fragments)
- (*c->returned_fragments)++;
- }
- }
-
-
- static int
- fragment_select (struct Plugin *plugin,
- const char *stmt,
- struct GNUNET_PQ_QueryParam *params,
- uint64_t *returned_fragments,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- /* Stack based closure */
- struct FragmentRowsContext frc = {
- .cb = cb,
- .cb_cls = cb_cls,
- .returned_fragments = returned_fragments,
- .ret = GNUNET_SYSERR
- };
-
- if (0 > GNUNET_PQ_eval_prepared_multi_select (plugin->dbh,
- stmt, params,
- &fragment_rows, &frc))
- return GNUNET_SYSERR;
- return frc.ret; /* GNUNET_OK ?? */
- }
-
- /**
- * Retrieve a message fragment range by fragment ID.
- *
- * @see GNUNET_PSYCSTORE_fragment_get()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- fragment_get (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t first_fragment_id,
- uint64_t last_fragment_id,
- uint64_t *returned_fragments,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&first_fragment_id),
- GNUNET_PQ_query_param_uint64 (&last_fragment_id),
- GNUNET_PQ_query_param_end
- };
-
- *returned_fragments = 0;
- return fragment_select (plugin,
- "select_fragments",
- params_select,
- returned_fragments,
- cb, cb_cls);
- }
-
-
- /**
- * Retrieve a message fragment range by fragment ID.
- *
- * @see GNUNET_PSYCSTORE_fragment_get_latest()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- fragment_get_latest (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t fragment_limit,
- uint64_t *returned_fragments,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
-
- *returned_fragments = 0;
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&fragment_limit),
- GNUNET_PQ_query_param_end
- };
-
- return fragment_select (plugin,
- "select_latest_fragments",
- params_select,
- returned_fragments,
- cb, cb_cls);
- }
-
-
- /**
- * Retrieve all fragments of a message ID range.
- *
- * @see GNUNET_PSYCSTORE_message_get()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- message_get (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t first_message_id,
- uint64_t last_message_id,
- uint64_t fragment_limit,
- uint64_t *returned_fragments,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&first_message_id),
- GNUNET_PQ_query_param_uint64 (&last_message_id),
- GNUNET_PQ_query_param_uint64 (&fragment_limit),
- GNUNET_PQ_query_param_end
- };
-
- if (0 == fragment_limit)
- fragment_limit = INT64_MAX;
- *returned_fragments = 0;
- return fragment_select (plugin,
- "select_messages",
- params_select,
- returned_fragments,
- cb, cb_cls);
- }
-
-
- /**
- * Retrieve all fragments of the latest messages.
- *
- * @see GNUNET_PSYCSTORE_message_get_latest()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- message_get_latest (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_limit,
- uint64_t *returned_fragments,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&message_limit),
- GNUNET_PQ_query_param_end
- };
-
- *returned_fragments = 0;
- return fragment_select (plugin,
- "select_latest_messages",
- params_select,
- returned_fragments,
- cb, cb_cls);
- }
-
-
- /**
- * Retrieve a fragment of message specified by its message ID and fragment
- * offset.
- *
- * @see GNUNET_PSYCSTORE_message_get_fragment()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- message_get_fragment (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_id,
- uint64_t fragment_offset,
- GNUNET_PSYCSTORE_FragmentCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
- const char *stmt = "select_message_fragment";
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_uint64 (&message_id),
- GNUNET_PQ_query_param_uint64 (&fragment_offset),
- GNUNET_PQ_query_param_end
- };
-
- /* Stack based closure */
- struct FragmentRowsContext frc = {
- .cb = cb,
- .cb_cls = cb_cls,
- .returned_fragments = NULL,
- .ret = GNUNET_SYSERR
- };
-
- if (0 > GNUNET_PQ_eval_prepared_multi_select (plugin->dbh,
- stmt, params_select,
- &fragment_rows, &frc))
- return GNUNET_SYSERR;
- return frc.ret; /* GNUNET_OK ?? */
- }
-
- /**
- * Retrieve the max. values of message counters for a channel.
- *
- * @see GNUNET_PSYCSTORE_counters_get()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- counters_message_get (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t *max_fragment_id,
- uint64_t *max_message_id,
- uint64_t *max_group_generation)
- {
- struct Plugin *plugin = cls;
-
- const char *stmt = "select_counters_message";
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- struct GNUNET_PQ_ResultSpec results_select[] = {
- GNUNET_PQ_result_spec_uint64 ("fragment_id", max_fragment_id),
- GNUNET_PQ_result_spec_uint64 ("message_id", max_message_id),
- GNUNET_PQ_result_spec_uint64 ("group_generation", max_group_generation),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_singleton_select (plugin->dbh, stmt,
- params_select, results_select))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
- /**
- * Retrieve the max. values of state counters for a channel.
- *
- * @see GNUNET_PSYCSTORE_counters_get()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- counters_state_get (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t *max_state_message_id)
- {
- struct Plugin *plugin = cls;
-
- const char *stmt = "select_counters_state";
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- struct GNUNET_PQ_ResultSpec results_select[] = {
- GNUNET_PQ_result_spec_uint64 ("max_state_message_id", max_state_message_id),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_singleton_select (plugin->dbh, stmt,
- params_select, results_select))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- /**
- * Assign a value to a state variable.
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_assign (struct Plugin *plugin, const char *stmt,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const char *name, const void *value, size_t value_size)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_string (name),
- GNUNET_PQ_query_param_fixed_size (value, value_size),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, stmt, params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- static int
- update_message_id (struct Plugin *plugin,
- const char *stmt,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_id)
- {
- struct GNUNET_PQ_QueryParam params[] = {
- GNUNET_PQ_query_param_uint64 (&message_id),
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS !=
- GNUNET_PQ_eval_prepared_non_select (plugin->dbh, stmt, params))
- return GNUNET_SYSERR;
-
- return GNUNET_OK;
- }
-
-
- /**
- * Begin modifying current state.
- */
- static int
- state_modify_begin (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_id, uint64_t state_delta)
- {
- struct Plugin *plugin = cls;
-
- if (state_delta > 0)
- {
- /**
- * We can only apply state modifiers in the current message if modifiers in
- * the previous stateful message (message_id - state_delta) were already
- * applied.
- */
-
- uint64_t max_state_message_id = 0;
- int ret = counters_state_get (plugin, channel_key, &max_state_message_id);
- switch (ret)
- {
- case GNUNET_OK:
- case GNUNET_NO: // no state yet
- ret = GNUNET_OK;
- break;
-
- default:
- return ret;
- }
-
- if (max_state_message_id < message_id - state_delta)
- return GNUNET_NO; /* some stateful messages not yet applied */
- else if (message_id - state_delta < max_state_message_id)
- return GNUNET_NO; /* changes already applied */
- }
-
- if (TRANSACTION_NONE != plugin->transaction)
- {
- /** @todo FIXME: wait for other transaction to finish */
- return GNUNET_SYSERR;
- }
- return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
- }
-
-
- /**
- * Set the current value of state variable.
- *
- * @see GNUNET_PSYCSTORE_state_modify()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_modify_op (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- enum GNUNET_PSYC_Operator op,
- const char *name, const void *value, size_t value_size)
- {
- struct Plugin *plugin = cls;
- GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
-
- switch (op)
- {
- case GNUNET_PSYC_OP_ASSIGN:
- return state_assign (plugin, "insert_state_current",
- channel_key, name, value, value_size);
-
- default: /** @todo implement more state operations */
- GNUNET_break (0);
- return GNUNET_SYSERR;
- }
- }
-
-
- /**
- * End modifying current state.
- */
- static int
- state_modify_end (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t message_id)
- {
- struct Plugin *plugin = cls;
- GNUNET_assert (TRANSACTION_STATE_MODIFY == plugin->transaction);
-
- return
- GNUNET_OK == exec_channel (plugin, "delete_state_empty", channel_key)
- && GNUNET_OK == update_message_id (plugin,
- "update_max_state_message_id",
- channel_key, message_id)
- && GNUNET_OK == transaction_commit (plugin)
- ? GNUNET_OK : GNUNET_SYSERR;
- }
-
-
- /**
- * Begin state synchronization.
- */
- static int
- state_sync_begin (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
- {
- struct Plugin *plugin = cls;
- return exec_channel (plugin, "delete_state_sync", channel_key);
- }
-
-
- /**
- * Assign current value of a state variable.
- *
- * @see GNUNET_PSYCSTORE_state_modify()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_sync_assign (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const char *name, const void *value, size_t value_size)
- {
- struct Plugin *plugin = cls;
- return state_assign (plugin, "insert_state_sync",
- channel_key, name, value, value_size);
- }
-
-
- /**
- * End modifying current state.
- */
- static int
- state_sync_end (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- uint64_t max_state_message_id,
- uint64_t state_hash_message_id)
- {
- struct Plugin *plugin = cls;
- int ret = GNUNET_SYSERR;
-
- if (TRANSACTION_NONE != plugin->transaction)
- {
- /** @todo FIXME: wait for other transaction to finish */
- return GNUNET_SYSERR;
- }
-
- GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC)
- && GNUNET_OK == exec_channel (plugin, "delete_state", channel_key)
- && GNUNET_OK == exec_channel (plugin, "insert_state_from_sync",
- channel_key)
- && GNUNET_OK == exec_channel (plugin, "delete_state_sync",
- channel_key)
- && GNUNET_OK == update_message_id (plugin,
- "update_state_hash_message_id",
- channel_key, state_hash_message_id)
- && GNUNET_OK == update_message_id (plugin,
- "update_max_state_message_id",
- channel_key, max_state_message_id)
- && GNUNET_OK == transaction_commit (plugin)
- ? ret = GNUNET_OK
- : transaction_rollback (plugin);
- return ret;
- }
-
-
- /**
- * Delete the whole state.
- *
- * @see GNUNET_PSYCSTORE_state_reset()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_reset (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
- {
- struct Plugin *plugin = cls;
- return exec_channel (plugin, "delete_state", channel_key);
- }
-
-
- /**
- * Update signed values of state variables in the state store.
- *
- * @see GNUNET_PSYCSTORE_state_hash_update()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_update_signed (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key)
- {
- struct Plugin *plugin = cls;
- return exec_channel (plugin, "update_state_signed", channel_key);
- }
-
-
- /**
- * Retrieve a state variable by name.
- *
- * @see GNUNET_PSYCSTORE_state_get()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_get (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const char *name, GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
- {
- struct Plugin *plugin = cls;
-
- const char *stmt = "select_state_one";
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_string (name),
- GNUNET_PQ_query_param_end
- };
-
- void *value_current = NULL;
- size_t value_size = 0;
-
- struct GNUNET_PQ_ResultSpec results_select[] = {
- GNUNET_PQ_result_spec_variable_size ("value_current", &value_current, &value_size),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_DB_STATUS_SUCCESS_ONE_RESULT !=
- GNUNET_PQ_eval_prepared_singleton_select (plugin->dbh, stmt,
- params_select, results_select))
- return GNUNET_SYSERR;
-
- return cb (cb_cls, name, value_current,
- value_size);
- }
-
-
-
- /**
- * Closure for #get_state_cb.
- */
- struct GetStateContext {
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key;
- // const char *name,
- GNUNET_PSYCSTORE_StateCallback cb;
- void *cb_cls;
-
- const char *value_id;
-
- /* I preserved this but I do not see the point since
- * it cannot stop the loop early and gets overwritten ?? */
- int ret;
- };
-
-
- /**
- * Callback that retrieves the results of a SELECT statement
- * reading form the state table.
- *
- * Only passed to GNUNET_PQ_eval_prepared_multi_select and
- * has type GNUNET_PQ_PostgresResultHandler.
- *
- * @param cls closure
- * @param result the postgres result
- * @param num_result the number of results in @a result
- */
- static void
- get_state_cb (void *cls,
- PGresult *res,
- unsigned int num_results)
- {
- struct GetStateContext *c = cls;
-
- for (unsigned int i=0;i<num_results;i++)
- {
- char *name = "";
- void *value = NULL;
- size_t value_size = 0;
-
- struct GNUNET_PQ_ResultSpec results[] = {
- GNUNET_PQ_result_spec_string ("name", &name),
- GNUNET_PQ_result_spec_variable_size (c->value_id, &value, &value_size),
- GNUNET_PQ_result_spec_end
- };
-
- if (GNUNET_YES != GNUNET_PQ_extract_result (res, results, i))
- {
- GNUNET_PQ_cleanup_result(results); /* previously invoked via PQclear?? */
- break; /* nothing more?? */
- }
-
- c->ret = c->cb (c->cb_cls, (const char *) name, value, value_size);
- GNUNET_PQ_cleanup_result(results);
- }
- }
-
- /**
- * Retrieve all state variables for a channel with the given prefix.
- *
- * @see GNUNET_PSYCSTORE_state_get_prefix()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- const char *name, GNUNET_PSYCSTORE_StateCallback cb,
- void *cb_cls)
- {
- struct Plugin *plugin = cls;
-
- const char *stmt = "select_state_prefix";
-
- uint32_t name_len = (uint32_t) strlen (name);
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_string (name),
- GNUNET_PQ_query_param_uint32 (&name_len),
- GNUNET_PQ_query_param_string (name),
- GNUNET_PQ_query_param_end
- };
-
- struct GetStateContext gsc = {
- .cb = cb,
- .cb_cls = cb_cls,
- .value_id = "value_current",
- .ret = GNUNET_NO
- };
-
- if (0 > GNUNET_PQ_eval_prepared_multi_select (plugin->dbh,
- stmt, params_select,
- &get_state_cb, &gsc))
- return GNUNET_SYSERR;
- return gsc.ret; /* GNUNET_OK ?? */
- }
-
-
- /**
- * Retrieve all signed state variables for a channel.
- *
- * @see GNUNET_PSYCSTORE_state_get_signed()
- *
- * @return #GNUNET_OK on success, else #GNUNET_SYSERR
- */
- static int
- state_get_signed (void *cls,
- const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
- GNUNET_PSYCSTORE_StateCallback cb, void *cb_cls)
- {
- struct Plugin *plugin = cls;
-
- const char *stmt = "select_state_signed";
-
- struct GNUNET_PQ_QueryParam params_select[] = {
- GNUNET_PQ_query_param_auto_from_type (channel_key),
- GNUNET_PQ_query_param_end
- };
-
- struct GetStateContext gsc = {
- .cb = cb,
- .cb_cls = cb_cls,
- .value_id = "value_signed",
- .ret = GNUNET_NO
- };
-
- if (0 > GNUNET_PQ_eval_prepared_multi_select (plugin->dbh,
- stmt, params_select,
- &get_state_cb, &gsc))
- return GNUNET_SYSERR;
- return gsc.ret; /* GNUNET_OK ?? */
- }
-
-
- /**
- * Entry point for the plugin.
- *
- * @param cls The struct GNUNET_CONFIGURATION_Handle.
- * @return NULL on error, otherwise the plugin context
- */
- void *
- libgnunet_plugin_psycstore_postgres_init (void *cls)
- {
- static struct Plugin plugin;
- const struct GNUNET_CONFIGURATION_Handle *cfg = cls;
- struct GNUNET_PSYCSTORE_PluginFunctions *api;
-
- if (NULL != plugin.cfg)
- return NULL; /* can only initialize once! */
- memset (&plugin, 0, sizeof (struct Plugin));
- plugin.cfg = cfg;
- if (GNUNET_OK != database_setup (&plugin))
- {
- database_shutdown (&plugin);
- return NULL;
- }
- api = GNUNET_new (struct GNUNET_PSYCSTORE_PluginFunctions);
- api->cls = &plugin;
- api->membership_store = &postgres_membership_store;
- api->membership_test = &membership_test;
- api->fragment_store = &fragment_store;
- api->message_add_flags = &message_add_flags;
- api->fragment_get = &fragment_get;
- api->fragment_get_latest = &fragment_get_latest;
- api->message_get = &message_get;
- api->message_get_latest = &message_get_latest;
- api->message_get_fragment = &message_get_fragment;
- api->counters_message_get = &counters_message_get;
- api->counters_state_get = &counters_state_get;
- api->state_modify_begin = &state_modify_begin;
- api->state_modify_op = &state_modify_op;
- api->state_modify_end = &state_modify_end;
- api->state_sync_begin = &state_sync_begin;
- api->state_sync_assign = &state_sync_assign;
- api->state_sync_end = &state_sync_end;
- api->state_reset = &state_reset;
- api->state_update_signed = &state_update_signed;
- api->state_get = &state_get;
- api->state_get_prefix = &state_get_prefix;
- api->state_get_signed = &state_get_signed;
-
- LOG (GNUNET_ERROR_TYPE_INFO, _("Postgres database running\n"));
- return api;
- }
-
-
- /**
- * Exit point from the plugin.
- *
- * @param cls The plugin context (as returned by "init")
- * @return Always NULL
- */
- void *
- libgnunet_plugin_psycstore_postgres_done (void *cls)
- {
- struct GNUNET_PSYCSTORE_PluginFunctions *api = cls;
- struct Plugin *plugin = api->cls;
-
- database_shutdown (plugin);
- plugin->cfg = NULL;
- GNUNET_free (api);
- LOG (GNUNET_ERROR_TYPE_DEBUG, "Postgres plugin has finished\n");
- return NULL;
- }
-
- /* end of plugin_psycstore_postgres.c */
|