b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "lib.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "array.h"

b87761f9bbef949f31dae297e619ac3f5e9c2b2eTimo Sirainen#include "hex-binary.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "str.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "ioloop.h"

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz#include "net.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "write-full.h"

de5f478d9e7ae7b8e58082e0b30b6ce1f034236aTimo Sirainen#include "time-util.h"

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen#include "var-expand.h"

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen#include "settings-parser.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include "sql-api-private.h"

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#ifdef BUILD_CASSANDRA

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen#include <fcntl.h>

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include <unistd.h>

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#include <cassandra.h>

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen#define IS_CONNECTED(db) \

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ((db)->api.state != SQL_DB_STATE_DISCONNECTED && \

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen (db)->api.state != SQL_DB_STATE_CONNECTING)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen#define CASSANDRA_FALLBACK_WARN_INTERVAL_SECS 60

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen#define CASSANDRA_FALLBACK_FIRST_RETRY_MSECS 50

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen#define CASSANDRA_FALLBACK_MAX_RETRY_MSECS (1000*60)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen#define CASS_QUERY_DEFAULT_WARN_TIMEOUT_MSECS (5*1000)

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainentypedef void driver_cassandra_callback_t(CassFuture *future, void *context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainenenum cassandra_query_type {

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen CASSANDRA_QUERY_TYPE_READ,

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen CASSANDRA_QUERY_TYPE_WRITE,

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen CASSANDRA_QUERY_TYPE_DELETE

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen};

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen#define CASSANDRA_QUERY_TYPE_COUNT 3

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainenstatic const char *cassandra_query_type_names[CASSANDRA_QUERY_TYPE_COUNT] = {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen "read", "write", "delete"

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen};

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstruct cassandra_callback {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int id;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassFuture *future;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_callback_t *callback;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen};

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstruct cassandra_db {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_db api;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen char *hosts, *keyspace, *user, *password;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen CassConsistency read_consistency, write_consistency, delete_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen CassConsistency read_fallback_consistency, write_fallback_consistency, delete_fallback_consistency;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen CassLogLevel log_level;

14189e0d0af45ddcb888d026bd8d7e4609912ec5Timo Sirainen bool debug_queries;

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen unsigned int protocol_version;

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen unsigned int num_threads;

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen unsigned int connect_timeout_msecs, request_timeout_msecs;

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen unsigned int warn_timeout_msecs;

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz in_port_t port;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassCluster *cluster;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassSession *session;

f0e416aa42058e7ccc0dc6deec0d4f4a19ee6ebeTimo Sirainen CassTimestampGen *timestamp_gen;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen int fd_pipe[2];

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct io *io_pipe;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ARRAY(struct cassandra_callback *) callbacks;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen ARRAY(struct cassandra_result *) results;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int callback_ids;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen char *metrics_path;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen struct timeout *to_metrics;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct timeval first_fallback_sent[CASSANDRA_QUERY_TYPE_COUNT];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen time_t last_fallback_warning[CASSANDRA_QUERY_TYPE_COUNT];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen unsigned int fallback_failures[CASSANDRA_QUERY_TYPE_COUNT];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen /* for synchronous queries: */

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct ioloop *ioloop, *orig_ioloop;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result *sync_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen char *error;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen};

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstruct cassandra_result {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result api;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassStatement *statement;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const CassResult *result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassIterator *iterator;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen char *query;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen char *error;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen CassConsistency consistency, fallback_consistency;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen enum cassandra_query_type query_type;

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen struct timeval start_time, finish_time;

37e8420b32a0fa3442c405616980e45beb494104Timo Sirainen unsigned int row_count;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen pool_t row_pool;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ARRAY_TYPE(const_string) fields;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen ARRAY(size_t) field_sizes;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_query_callback_t *callback;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

0dffa25d211be541ee3c953b23566a1a990789dfTimo Sirainen bool query_sent:1;

0dffa25d211be541ee3c953b23566a1a990789dfTimo Sirainen bool finished:1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen};

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstruct cassandra_transaction_context {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_transaction_context ctx;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen int refcount;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_commit_callback_t *callback;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen pool_t query_pool;

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen char *error;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

0dffa25d211be541ee3c953b23566a1a990789dfTimo Sirainen bool begin_succeeded:1;

0dffa25d211be541ee3c953b23566a1a990789dfTimo Sirainen bool begin_failed:1;

0dffa25d211be541ee3c953b23566a1a990789dfTimo Sirainen bool failed:1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen};

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenextern const struct sql_db driver_cassandra_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenextern const struct sql_result driver_cassandra_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic struct {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassConsistency consistency;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *name;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen} cass_consistency_names[] = {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_ANY, "any" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_ONE, "one" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_TWO, "two" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_THREE, "three" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_QUORUM, "" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_ALL, "all" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_QUORUM, "" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_ALL, "all" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_LOCAL_QUORUM, "local-quorum" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_EACH_QUORUM, "each-quorum" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_SERIAL, "serial" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_LOCAL_SERIAL, "local-serial" },

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen { CASS_CONSISTENCY_LOCAL_ONE, "local-one" }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen};

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainenstatic struct {

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen CassLogLevel log_level;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen const char *name;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen} cass_log_level_names[] = {

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_CRITICAL, "critical" },

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_ERROR, "error" },

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_WARN, "warn" },

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_INFO, "info" },

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_DEBUG, "debug" },

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen { CASS_LOG_TRACE, "trace" }

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen};

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainenstatic void driver_cassandra_result_send_query(struct cassandra_result *result);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainenstatic void driver_cassandra_send_queries(struct cassandra_db *db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void result_finish(struct cassandra_result *result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic int consistency_parse(const char *str, CassConsistency *consistency_r)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int i;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen for (i = 0; i < N_ELEMENTS(cass_consistency_names); i++) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (strcmp(cass_consistency_names[i].name, str) == 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen *consistency_r = cass_consistency_names[i].consistency;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return 0;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainenstatic int log_level_parse(const char *str, CassLogLevel *log_level_r)

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen{

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen unsigned int i;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen for (i = 0; i < N_ELEMENTS(cass_log_level_names); i++) {

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen if (strcmp(cass_log_level_names[i].name, str) == 0) {

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen *log_level_r = cass_log_level_names[i].log_level;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen return 0;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen }

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen }

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen return -1;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen}

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_set_state(struct cassandra_db *db, enum sql_db_state state)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* switch back to original ioloop in case the caller wants to

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->ioloop != NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_set_current(db->orig_ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_db_set_state(&db->api, state);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->ioloop != NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_set_current(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainenstatic void driver_cassandra_close(struct cassandra_db *db, const char *error)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen struct cassandra_result *const *resultp;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->io_pipe != NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_remove(&db->io_pipe);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->fd_pipe[0] != -1) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_close_fd(&db->fd_pipe[0]);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_close_fd(&db->fd_pipe[1]);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_DISCONNECTED);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen while (array_count(&db->results) > 0) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen resultp = array_idx(&db->results, 0);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if ((*resultp)->error == NULL)

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen (*resultp)->error = i_strdup(error);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen result_finish(*resultp);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->ioloop != NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* running a sync query, stop it */

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_stop(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_log_error(CassFuture *future, const char *str)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *message;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen size_t size;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_future_error_message(future, &message, &size);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("cassandra: %s: %.*s", str, (int)size, message);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_future_callback(CassFuture *future ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_callback *cb = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* this isn't the main thread - communicate with main thread by

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (write_full(cb->db->fd_pipe[1], &cb->id, sizeof(cb->id)) < 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("cassandra: write(pipe) failed: %m");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void cassandra_callback_run(struct cassandra_callback *cb)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->callback(cb->future, cb->context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_future_free(cb->future);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(cb);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_input_id(struct cassandra_db *db, unsigned int id)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_callback *cb, *const *cbp;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* usually there are only a few callbacks, so don't bother with using

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_foreach(&db->callbacks, cbp) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb = *cbp;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (cb->id == id) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_delete(&db->callbacks,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_foreach_idx(&db->callbacks, cbp), 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cassandra_callback_run(cb);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_panic("cassandra: Received unknown ID %u", id);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_input(struct cassandra_db *db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int ids[1024];

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ssize_t ret;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ret = read(db->fd_pipe[0], ids, sizeof(ids));

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (ret < 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("cassandra: read(pipe) failed: %m");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen else if (ret == 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("cassandra: read(pipe) failed: EOF");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen else if (ret % sizeof(ids[0]) != 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("cassandra: read(pipe) returned wrong amount of data");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen else {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* success */

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int i, count = ret / sizeof(ids[0]);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen for (i = 0; i < count && db->api.state != SQL_DB_STATE_DISCONNECTED; i++)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_input_id(db, ids[i]);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_close(db, "IPC pipe closed");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_set_callback(CassFuture *future, struct cassandra_db *db,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_callback_t *callback,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_callback *cb;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb = i_new(struct cassandra_callback, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->id = ++db->callback_ids;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->future = future;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->callback = callback;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->context = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cb->db = db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_append(&db->callbacks, &cb, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_future_set_callback(future, driver_cassandra_future_callback, cb);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void connect_callback(CassFuture *future, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassError rc;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if ((rc = cass_future_error_code(future)) != CASS_OK) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_log_error(future,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen "Couldn't connect to Cassandra");

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_close(db, "Couldn't connect to Cassandra");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_IDLE);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->ioloop != NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* driver_cassandra_sync_init() waiting for connection to

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_stop(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_send_queries(db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic int driver_cassandra_connect(struct sql_db *_db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassFuture *future;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(db->api.state == SQL_DB_STATE_DISCONNECTED);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (pipe(db->fd_pipe) < 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_error("pipe() failed: %m");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->io_pipe = io_add(db->fd_pipe[0], IO_READ,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_input, db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_set_state(db, SQL_DB_STATE_CONNECTING);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen future = cass_session_connect_keyspace(db->session, db->cluster, db->keyspace);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_set_callback(future, db, connect_callback, db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return 0;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_disconnect(struct sql_db *_db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_close(db, "Disconnected");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_escape_string(struct sql_db *db ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *string)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen string_t *escaped;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int i;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (strchr(string, '\'') == NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return string;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen escaped = t_str_new(strlen(string)+10);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen for (i = 0; string[i] != '\0'; i++) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (string[i] == '\'')

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen str_append_c(escaped, '\'');

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen str_append_c(escaped, string[i]);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return str_c(escaped);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_parse_connect_string(struct cassandra_db *db,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *connect_string)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen const char *const *args, *key, *value, *error;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen string_t *hosts = t_str_new(64);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen bool read_fallback_set = FALSE, write_fallback_set = FALSE, delete_fallback_set = FALSE;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen db->log_level = CASS_LOG_WARN;

6ea145a99eeee923602f04d3c9183bbdba6cd190Timo Sirainen db->read_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

6ea145a99eeee923602f04d3c9183bbdba6cd190Timo Sirainen db->write_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen db->delete_consistency = CASS_CONSISTENCY_LOCAL_QUORUM;

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen db->connect_timeout_msecs = SQL_CONNECT_TIMEOUT_SECS*1000;

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen db->request_timeout_msecs = SQL_QUERY_TIMEOUT_SECS*1000;

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen db->warn_timeout_msecs = CASS_QUERY_DEFAULT_WARN_TIMEOUT_MSECS;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen args = t_strsplit_spaces(connect_string, " ");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen for (; *args != NULL; args++) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen value = strchr(*args, '=');

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (value == NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_fatal("cassandra: Missing value in connect string: %s",

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen *args);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen key = t_strdup_until(*args, value++);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (strcmp(key, "host") == 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (str_len(hosts) > 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen str_append_c(hosts, ',');

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen str_append(hosts, value);

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz } else if (strcmp(key, "port") == 0) {

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz if (net_str2port(value, &db->port) < 0)

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz i_fatal("cassandra: Invalid port: %s", value);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } else if (strcmp(key, "dbname") == 0 ||

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen strcmp(key, "keyspace") == 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(db->keyspace);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->keyspace = i_strdup(value);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen } else if (strcmp(key, "user") == 0) {

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen i_free(db->user);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen db->user = i_strdup(value);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen } else if (strcmp(key, "password") == 0) {

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen i_free(db->password);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen db->password = i_strdup(value);

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen } else if (strcmp(key, "read_consistency") == 0) {

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen if (consistency_parse(value, &db->read_consistency) < 0)

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen i_fatal("cassandra: Unknown read_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen } else if (strcmp(key, "read_fallback_consistency") == 0) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (consistency_parse(value, &db->read_fallback_consistency) < 0)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen i_fatal("cassandra: Unknown read_fallback_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen read_fallback_set = TRUE;

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen } else if (strcmp(key, "write_consistency") == 0) {

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen if (consistency_parse(value, &db->write_consistency) < 0)

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen i_fatal("cassandra: Unknown write_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen } else if (strcmp(key, "write_fallback_consistency") == 0) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (consistency_parse(value, &db->write_fallback_consistency) < 0)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen i_fatal("cassandra: Unknown write_fallback_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen write_fallback_set = TRUE;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen } else if (strcmp(key, "delete_consistency") == 0) {

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen if (consistency_parse(value, &db->delete_consistency) < 0)

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen i_fatal("cassandra: Unknown delete_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen } else if (strcmp(key, "delete_fallback_consistency") == 0) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (consistency_parse(value, &db->delete_fallback_consistency) < 0)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen i_fatal("cassandra: Unknown delete_fallback_consistency: %s", value);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen delete_fallback_set = TRUE;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen } else if (strcmp(key, "log_level") == 0) {

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen if (log_level_parse(value, &db->log_level) < 0)

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen i_fatal("cassandra: Unknown log_level: %s", value);

14189e0d0af45ddcb888d026bd8d7e4609912ec5Timo Sirainen } else if (strcmp(key, "debug_queries") == 0) {

14189e0d0af45ddcb888d026bd8d7e4609912ec5Timo Sirainen db->debug_queries = TRUE;

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen } else if (strcmp(key, "version") == 0) {

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen if (str_to_uint(value, &db->protocol_version) < 0)

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen i_fatal("cassandra: Invalid version: %s", value);

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen } else if (strcmp(key, "num_threads") == 0) {

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen if (str_to_uint(value, &db->num_threads) < 0)

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen i_fatal("cassandra: Invalid num_threads: %s", value);

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen } else if (strcmp(key, "connect_timeout") == 0) {

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen if (settings_get_time_msecs(value, &db->connect_timeout_msecs, &error) < 0)

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen i_fatal("cassandra: Invalid connect_timeout '%s': %s", value, error);

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen } else if (strcmp(key, "request_timeout") == 0) {

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen if (settings_get_time_msecs(value, &db->request_timeout_msecs, &error) < 0)

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen i_fatal("cassandra: Invalid request_timeout '%s': %s", value, error);

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen } else if (strcmp(key, "warn_timeout") == 0) {

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen if (settings_get_time_msecs(value, &db->warn_timeout_msecs, &error) < 0)

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen i_fatal("cassandra: Invalid warn_timeout '%s': %s", value, error);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen } else if (strcmp(key, "metrics") == 0) {

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen i_free(db->metrics_path);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen db->metrics_path = i_strdup(value);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } else {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_fatal("cassandra: Unknown connect string: %s", key);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (!read_fallback_set)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->read_fallback_consistency = db->read_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (!write_fallback_set)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->write_fallback_consistency = db->write_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (!delete_fallback_set)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->delete_fallback_consistency = db->delete_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (str_len(hosts) == 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_fatal("cassandra: No hosts given in connect string");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->keyspace == NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_fatal("cassandra: No dbname given in connect string");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->hosts = i_strdup(str_c(hosts));

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainenstatic void

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainendriver_cassandra_get_metrics_json(struct cassandra_db *db, string_t *dest)

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen{

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen#define ADD_UINT64(_struct, _field) \

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_printfa(dest, "\""#_field"\": %llu,", (unsigned long long)metrics._struct._field);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen#define ADD_DOUBLE(_struct, _field) \

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_printfa(dest, "\""#_field"\": %02lf,", metrics._struct._field);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen CassMetrics metrics;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen cass_session_get_metrics(db->session, &metrics);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_append(dest, "{ \"requests\": {");

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, min);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, max);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, mean);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, stddev);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, median);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, percentile_75th);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, percentile_95th);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, percentile_98th);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, percentile_99th);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(requests, percentile_999th);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_DOUBLE(requests, mean_rate);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_DOUBLE(requests, one_minute_rate);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_DOUBLE(requests, five_minute_rate);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_DOUBLE(requests, fifteen_minute_rate);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_truncate(dest, str_len(dest)-1);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_append(dest, "}, \"stats\": {");

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(stats, total_connections);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(stats, available_connections);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(stats, exceeded_pending_requests_water_mark);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(stats, exceeded_write_bytes_water_mark);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_truncate(dest, str_len(dest)-1);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_append(dest, "}, \"errors\": {");

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(errors, connection_timeouts);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(errors, pending_request_timeouts);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen ADD_UINT64(errors, request_timeouts);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_truncate(dest, str_len(dest)-1);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen str_append(dest, "}}");

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen}

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainenstatic void driver_cassandra_metrics_write(struct cassandra_db *db)

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen{

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen struct var_expand_table tab[] = {

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen { '\0', NULL, NULL }

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen };

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen string_t *path = t_str_new(64);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen string_t *data;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen int fd;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen var_expand(path, db->metrics_path, tab);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen fd = open(str_c(path), O_WRONLY | O_CREAT | O_TRUNC | O_NONBLOCK, 0600);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen if (fd == -1) {

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen i_error("creat(%s) failed: %m", str_c(path));

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen return;

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen }

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen data = t_str_new(1024);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen driver_cassandra_get_metrics_json(db, data);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen if (write_full(fd, str_data(data), str_len(data)) < 0)

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen i_error("write(%s) failed: %m", str_c(path));

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen i_close_fd(&fd);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen}

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic struct sql_db *driver_cassandra_init_v(const char *connect_string)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db = i_new(struct cassandra_db, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->api = driver_cassandra_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->fd_pipe[0] = db->fd_pipe[1] = -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen T_BEGIN {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_parse_connect_string(db, connect_string);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } T_END;

a1044a46a8f3512173f4ea2684ef1fc3e61645c7Timo Sirainen cass_log_set_level(db->log_level);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

be59f9ae981dbe4bdd264053e9febd4ea5dad75bTimo Sirainen db->timestamp_gen = cass_timestamp_gen_monotonic_new();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->cluster = cass_cluster_new();

f0e416aa42058e7ccc0dc6deec0d4f4a19ee6ebeTimo Sirainen cass_cluster_set_timestamp_gen(db->cluster, db->timestamp_gen);

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen cass_cluster_set_connect_timeout(db->cluster, db->connect_timeout_msecs);

7cd055a212d44067e2d94452c05691d696c9f699Timo Sirainen cass_cluster_set_request_timeout(db->cluster, db->request_timeout_msecs);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_cluster_set_contact_points(db->cluster, db->hosts);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen if (db->user != NULL && db->password != NULL)

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen cass_cluster_set_credentials(db->cluster, db->user, db->password);

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz if (db->port != 0)

e8434aad92ea6ff1c915b708294dbd0c7ff5908dMichael M Slusarz cass_cluster_set_port(db->cluster, db->port);

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen if (db->protocol_version != 0)

f9eee365367f37b1692c07db6c23d30243844aaaTimo Sirainen cass_cluster_set_protocol_version(db->cluster, db->protocol_version);

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen if (db->num_threads != 0)

f7d018e7e0980044e3d537958126e44ef4c45056Timo Sirainen cass_cluster_set_num_threads_io(db->cluster, db->num_threads);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->session = cass_session_new();

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen if (db->metrics_path != NULL)

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen db->to_metrics = timeout_add(1000, driver_cassandra_metrics_write, db);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen i_array_init(&db->results, 16);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_array_init(&db->callbacks, 16);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return &db->api;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_deinit_v(struct sql_db *_db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_close(db, "Deinitialized");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(array_count(&db->callbacks) == 0);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_free(&db->callbacks);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen i_assert(array_count(&db->results) == 0);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen array_free(&db->results);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_session_free(db->session);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_cluster_free(db->cluster);

f0e416aa42058e7ccc0dc6deec0d4f4a19ee6ebeTimo Sirainen cass_timestamp_gen_free(db->timestamp_gen);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen if (db->to_metrics != NULL)

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen timeout_remove(&db->to_metrics);

ba4626cd5be3d225a7a89aa338d92b8fb411fd1cTimo Sirainen i_free(db->metrics_path);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(db->hosts);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(db->error);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(db->keyspace);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen i_free(db->user);

c72cfe4a2bda39fff3b8a8bd64b31a7cc14d7d11Timo Sirainen i_free(db->password);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_free(&_db->module_contexts);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainenstatic void driver_cassandra_result_unlink(struct cassandra_db *db,

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen struct cassandra_result *result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen struct cassandra_result *const *results;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen unsigned int i, count;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen results = array_get(&db->results, &count);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen for (i = 0; i < count; i++) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if (results[i] == result) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen array_delete(&db->results, i, 1);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen return;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen i_unreached();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_result_free(struct sql_result *_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_result->db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen struct timeval now;

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen const char *str;

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen long long reply_usecs;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b457d2ecf97fb52064f9dd563fd4e8065af39dfbTimo Sirainen i_assert(!result->api.callback);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(result->callback == NULL);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (_result == db->sync_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->sync_result = NULL;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen reply_usecs = timeval_diff_usecs(&result->finish_time, &result->start_time);

14189e0d0af45ddcb888d026bd8d7e4609912ec5Timo Sirainen if (db->log_level >= CASS_LOG_DEBUG || db->debug_queries ||

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen reply_usecs/1000 >= db->warn_timeout_msecs) {

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen if (gettimeofday(&now, NULL) < 0)

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen i_fatal("gettimeofday() failed: %m");

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen str = t_strdup_printf(

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen "cassandra: Finished query '%s' (%u rows, %lld+%lld us): %s",

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen result->query, result->row_count, reply_usecs,

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen timeval_diff_usecs(&now, &result->finish_time),

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen result->error != NULL ? result->error : "success");

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen if (reply_usecs/1000 >= db->warn_timeout_msecs)

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen i_warning("%s", str);

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen else

8952d797eca36f997ec36569e783871b597a9216Timo Sirainen i_debug("%s", str);

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen }

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen

f0e811f0e306bb20d3da9c26353bdd5669132f29Timo Sirainen if (result->result != NULL)

f0e811f0e306bb20d3da9c26353bdd5669132f29Timo Sirainen cass_result_free(result->result);

f0e811f0e306bb20d3da9c26353bdd5669132f29Timo Sirainen if (result->iterator != NULL)

f0e811f0e306bb20d3da9c26353bdd5669132f29Timo Sirainen cass_iterator_free(result->iterator);

3e8842470a4a17017529d43b39c40a7549c2ecf2Timo Sirainen if (result->statement != NULL)

3e8842470a4a17017529d43b39c40a7549c2ecf2Timo Sirainen cass_statement_free(result->statement);

3e8842470a4a17017529d43b39c40a7549c2ecf2Timo Sirainen if (result->row_pool != NULL)

3e8842470a4a17017529d43b39c40a7549c2ecf2Timo Sirainen pool_unref(&result->row_pool);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(result->query);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(result->error);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void result_finish(struct cassandra_result *result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen bool free_result = TRUE;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->finished = TRUE;

2ccb478c35972517721ce415d81fcbd11a73fad3Timo Sirainen result->finish_time = ioloop_timeval;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_result_unlink(db, result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainen i_assert((result->error != NULL) == (result->iterator == NULL));

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->api.callback = TRUE;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen T_BEGIN {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->callback(&result->api, result->context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } T_END;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->api.callback = FALSE;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen free_result = db->sync_result != &result->api;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->ioloop != NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_stop(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1856c361aad526948d56d8aafd576bca94516b92Timo Sirainen i_assert(!free_result || result->api.refcount > 0);

1856c361aad526948d56d8aafd576bca94516b92Timo Sirainen result->callback = NULL;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (free_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_result_unref(&result->api);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainenstatic void query_resend_with_fallback(struct cassandra_result *result)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen{

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen time_t last_warning =

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen ioloop_time - db->last_fallback_warning[result->query_type];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (last_warning >= CASSANDRA_FALLBACK_WARN_INTERVAL_SECS) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen i_warning("%s - retrying future %s queries with consistency %s (instead of %s)",

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->error, cassandra_query_type_names[result->query_type],

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen cass_consistency_string(result->fallback_consistency),

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen cass_consistency_string(result->consistency));

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->last_fallback_warning[result->query_type] = ioloop_time;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen i_free_and_null(result->error);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (db->fallback_failures[result->query_type]++ == 0)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->first_fallback_sent[result->query_type] = ioloop_timeval;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->consistency = result->fallback_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen driver_cassandra_result_send_query(result);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen}

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void query_callback(CassFuture *future, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = context;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen CassError error = cass_future_error_code(future);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (error != CASS_OK) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *errmsg;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen size_t errsize;

e401fa68eb1e7761ffd0b747919d44568555efeeTimo Sirainen int msecs;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen cass_future_error_message(future, &errmsg, &errsize);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(result->error);

e401fa68eb1e7761ffd0b747919d44568555efeeTimo Sirainen

e401fa68eb1e7761ffd0b747919d44568555efeeTimo Sirainen msecs = timeval_diff_msecs(&ioloop_timeval, &result->start_time);

e401fa68eb1e7761ffd0b747919d44568555efeeTimo Sirainen result->error = i_strdup_printf("Query '%s' failed: %.*s (in %u.%03u secs)",

e401fa68eb1e7761ffd0b747919d44568555efeeTimo Sirainen result->query, (int)errsize, errmsg, msecs/1000, msecs%1000);

2522acb523343dd37bf788747d86d9470fc08025Timo Sirainen /* unavailable = cassandra server knows that there aren't

2522acb523343dd37bf788747d86d9470fc08025Timo Sirainen if ((error == CASS_ERROR_SERVER_UNAVAILABLE ||

2522acb523343dd37bf788747d86d9470fc08025Timo Sirainen error == CASS_ERROR_SERVER_WRITE_TIMEOUT) &&

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->fallback_consistency != result->consistency) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen /* retry with fallback consistency */

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen query_resend_with_fallback(result);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen return;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result_finish(result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (result->fallback_consistency != result->consistency) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen /* non-fallback query finished successfully. if there had been

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen db->fallback_failures[result->query_type] = 0;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->result = cass_future_get_result(future);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->iterator = cass_iterator_from_result(result->result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result_finish(result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainenstatic void driver_cassandra_result_send_query(struct cassandra_result *result)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen{

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen CassFuture *future;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->statement = cass_statement_new(result->query, 0);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen cass_statement_set_consistency(result->statement, result->consistency);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen future = cass_session_execute(db->session, result->statement);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen driver_cassandra_set_callback(future, db, query_callback, result);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen}

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainenstatic bool

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainendriver_cassandra_want_fallback_query(struct cassandra_result *result)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen{

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen unsigned int failure_count = db->fallback_failures[result->query_type];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen unsigned int i, msecs = CASSANDRA_FALLBACK_FIRST_RETRY_MSECS;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen struct timeval tv;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (failure_count == 0)

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen return FALSE;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen tv = db->first_fallback_sent[result->query_type];

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen for (i = 1; i < failure_count; i++) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen msecs *= 2;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (msecs >= CASSANDRA_FALLBACK_MAX_RETRY_MSECS) {

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen msecs = CASSANDRA_FALLBACK_FIRST_RETRY_MSECS;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen break;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen timeval_add_msecs(&tv, msecs);

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen return timeval_cmp(&ioloop_timeval, &tv) < 0;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen}

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainenstatic int driver_cassandra_send_query(struct cassandra_result *result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)result->api.db;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen int ret;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if (!SQL_DB_IS_READY(&db->api)) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if ((ret = sql_connect(&db->api)) <= 0) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if (ret < 0)

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen driver_cassandra_close(db, "Couldn't connect to Cassandra");

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen return ret;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

de5f478d9e7ae7b8e58082e0b30b6ce1f034236aTimo Sirainen result->start_time = ioloop_timeval;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->row_pool = pool_alloconly_create("cassandra result", 512);

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen switch (result->query_type) {

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen case CASSANDRA_QUERY_TYPE_READ:

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->consistency = db->read_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->fallback_consistency = db->read_fallback_consistency;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen break;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen case CASSANDRA_QUERY_TYPE_WRITE:

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->consistency = db->write_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->fallback_consistency = db->write_fallback_consistency;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen break;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen case CASSANDRA_QUERY_TYPE_DELETE:

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->consistency = db->delete_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->fallback_consistency = db->delete_fallback_consistency;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen break;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen }

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen if (driver_cassandra_want_fallback_query(result))

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen result->consistency = result->fallback_consistency;

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen

1128c114416bdc4df0b41d3e15429a1522e5cfe4Timo Sirainen driver_cassandra_result_send_query(result);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen result->query_sent = TRUE;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen return 1;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen}

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainenstatic void driver_cassandra_send_queries(struct cassandra_db *db)

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen{

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen struct cassandra_result *const *results;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen unsigned int i, count;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen results = array_get(&db->results, &count);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen for (i = 0; i < count; i++) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if (!results[i]->query_sent) {

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen if (driver_cassandra_send_query(results[i]) <= 0)

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen break;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void exec_callback(struct sql_result *_result ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen void *context ATTR_UNUSED)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

ce74395e2a932342e04fb682395bcce111574969Timo Sirainenstatic void

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainendriver_cassandra_query_full(struct sql_db *_db, const char *query,

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen enum cassandra_query_type query_type,

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen sql_query_callback_t *callback, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result = i_new(struct cassandra_result, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->api = driver_cassandra_result;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen result->api.db = _db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->api.refcount = 1;

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen result->callback = callback;

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen result->context = context;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen result->query_type = query_type;

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen result->query = i_strdup(query);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen array_append(&db->results, &result, 1);

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen

1fb5e50695bbbc0da082e5a6f19f29d2bb2f6531Timo Sirainen (void)driver_cassandra_send_query(result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

ce74395e2a932342e04fb682395bcce111574969Timo Sirainenstatic void driver_cassandra_exec(struct sql_db *db, const char *query)

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen{

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen driver_cassandra_query_full(db, query, CASSANDRA_QUERY_TYPE_WRITE, exec_callback, NULL);

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen}

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_query(struct sql_db *db, const char *query,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_query_callback_t *callback, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen driver_cassandra_query_full(db, query, CASSANDRA_QUERY_TYPE_READ, callback, context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void cassandra_query_s_callback(struct sql_result *result, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->sync_result = result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_sync_init(struct cassandra_db *db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (sql_connect(&db->api) < 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->orig_ioloop = current_ioloop;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->ioloop = io_loop_create();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (IS_CONNECTED(db))

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(db->api.state == SQL_DB_STATE_CONNECTING);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* wait for connecting to finish */

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_run(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void driver_cassandra_sync_deinit(struct cassandra_db *db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->orig_ioloop == NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

e07677bb15404a3c18ad205efae86d6db31c3150Timo Sirainen if (db->io_pipe != NULL) {

e07677bb15404a3c18ad205efae86d6db31c3150Timo Sirainen io_loop_set_current(db->orig_ioloop);

e07677bb15404a3c18ad205efae86d6db31c3150Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

e07677bb15404a3c18ad205efae86d6db31c3150Timo Sirainen io_loop_set_current(db->ioloop);

e07677bb15404a3c18ad205efae86d6db31c3150Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_destroy(&db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic struct sql_result *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_sync_query(struct cassandra_db *db, const char *query)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result *result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(db->sync_result == NULL);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen switch (db->api.state) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen case SQL_DB_STATE_CONNECTING:

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen case SQL_DB_STATE_BUSY:

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_unreached();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen case SQL_DB_STATE_DISCONNECTED:

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_not_connected_result.refcount++;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return &sql_not_connected_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen case SQL_DB_STATE_IDLE:

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen break;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_query(&db->api, query, cassandra_query_s_callback, db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (db->sync_result == NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->io_pipe = io_loop_move_io(&db->io_pipe);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen io_loop_run(db->ioloop);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result = db->sync_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (result == &sql_not_connected_result) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* we don't end up in cassandra's free function, so sync_result

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen db->sync_result = NULL;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } else if (result == NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result = &sql_not_connected_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result->refcount++;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic struct sql_result *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_query_s(struct sql_db *_db, const char *query)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)_db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result *result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_sync_init(db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result = driver_cassandra_sync_query(db, query);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen driver_cassandra_sync_deinit(db);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic int

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_get_value(struct cassandra_result *result,

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen const CassValue *value, const char **str_r,

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen size_t *len_r)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen const unsigned char *output;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen void *output_dup;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen size_t output_size;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen CassError rc;

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen const char *type;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

23bdbb7b1831785c6ba6df190f6369da882d2b9dTimo Sirainen if (cass_value_is_null(value) != 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen *str_r = NULL;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return 0;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen switch (cass_data_type_type(cass_value_data_type(value))) {

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen case CASS_VALUE_TYPE_INT: {

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen cass_int32_t num;

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen rc = cass_value_get_int32(value, &num);

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen if (rc == CASS_OK) {

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen const char *str = t_strdup_printf("%d", num);

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen output_size = strlen(str);

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen output = (const void *)str;

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen }

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen type = "int32";

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen break;

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen }

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen default:

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen rc = cass_value_get_bytes(value, &output, &output_size);

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen type = "bytes";

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen break;

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (rc != CASS_OK) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(result->error);

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen result->error = i_strdup_printf("Couldn't get value as %s: %s",

b096ecf3188cdb9162460ed7ae885c03f3161462Timo Sirainen type, cass_error_desc(rc));

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen output_dup = p_malloc(result->row_pool, output_size + 1);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen memcpy(output_dup, output, output_size);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen *str_r = output_dup;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen *len_r = output_size;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return 0;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic int driver_cassandra_result_next_row(struct sql_result *_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const CassRow *row;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const CassValue *value;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *str;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen size_t size;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int i;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen int ret = 1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (result->iterator == NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

23bdbb7b1831785c6ba6df190f6369da882d2b9dTimo Sirainen if (cass_iterator_next(result->iterator) == 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return 0;

37e8420b32a0fa3442c405616980e45beb494104Timo Sirainen result->row_count++;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen p_clear(result->row_pool);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen p_array_init(&result->fields, result->row_pool, 8);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen p_array_init(&result->field_sizes, result->row_pool, 8);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen row = cass_iterator_get_row(result->iterator);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen for (i = 0; (value = cass_row_get_column(row, i)) != NULL; i++) {

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen if (driver_cassandra_get_value(result, value, &str, &size) < 0) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ret = -1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen break;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen array_append(&result->fields, &str, 1);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen array_append(&result->field_sizes, &size, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return ret;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic unsigned int

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_get_fields_count(struct sql_result *_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return array_count(&result->fields);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_get_field_name(struct sql_result *_result ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int idx ATTR_UNUSED)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_unreached();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic int

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_find_field(struct sql_result *_result ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *field_name ATTR_UNUSED)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_unreached();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_get_field_value(struct sql_result *_result,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int idx)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *const *strp;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen strp = array_idx(&result->fields, idx);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return *strp;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const unsigned char *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_get_field_value_binary(struct sql_result *_result ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen unsigned int idx ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen size_t *size_r ATTR_UNUSED)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen const char *const *strp;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen const size_t *sizep;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen strp = array_idx(&result->fields, idx);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen sizep = array_idx(&result->field_sizes, idx);

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen *size_r = *sizep;

61f39b0358a72ebc693d84ba5bac74489ee7df41Timo Sirainen return (const void *)*strp;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_find_field_value(struct sql_result *result ATTR_UNUSED,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen const char *field_name ATTR_UNUSED)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_unreached();

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *const *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_result_get_values(struct sql_result *_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return array_idx(&result->fields, 0);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic const char *driver_cassandra_result_get_error(struct sql_result *_result)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_result *result = (struct cassandra_result *)_result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (result->error != NULL)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return result->error;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return "FIXME";

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic struct sql_transaction_context *

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_transaction_begin(struct sql_db *db)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_transaction_context *ctx;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx = i_new(struct cassandra_transaction_context, 1);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->ctx.db = db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->refcount = 1;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* we need to be able to handle multiple open transactions, so at least

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->query_pool = pool_alloconly_create("cassandra transaction", 1024);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return &ctx->ctx;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainendriver_cassandra_transaction_unref(struct cassandra_transaction_context **_ctx)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainen struct cassandra_transaction_context *ctx = *_ctx;

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainen

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainen *_ctx = NULL;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_assert(ctx->refcount > 0);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (--ctx->refcount > 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen return;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen pool_unref(&ctx->query_pool);

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen i_free(ctx->error);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen i_free(ctx);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainenstatic void

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainentransaction_set_failed(struct cassandra_transaction_context *ctx,

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen const char *error)

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen{

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen if (ctx->failed) {

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen i_assert(ctx->error != NULL);

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen } else {

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen i_assert(ctx->error == NULL);

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen ctx->failed = TRUE;

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen ctx->error = i_strdup(error);

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen }

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen}

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void

ce74395e2a932342e04fb682395bcce111574969Timo Sirainentransaction_commit_callback(struct sql_result *result, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen struct cassandra_transaction_context *ctx = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (sql_result_next_row(result) < 0)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->callback(sql_result_get_error(result), ctx->context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen else

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->callback(NULL, ctx->context);

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainen driver_cassandra_transaction_unref(&ctx);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainendriver_cassandra_transaction_commit(struct sql_transaction_context *_ctx,

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen sql_commit_callback_t *callback, void *context)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_transaction_context *ctx =

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen (struct cassandra_transaction_context *)_ctx;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen enum cassandra_query_type query_type;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->callback = callback;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen ctx->context = context;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (ctx->failed || _ctx->head == NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen callback(ctx->failed ? ctx->error : NULL, context);

6b4b3e5fe8d9e84f4b1356ee898ca76996a11fe1Timo Sirainen driver_cassandra_transaction_unref(&ctx);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } else if (_ctx->head->next == NULL) {

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen /* just a single query, send it */

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen if (strncasecmp(_ctx->head->query, "DELETE ", 7) == 0)

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen query_type = CASSANDRA_QUERY_TYPE_DELETE;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen else

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen query_type = CASSANDRA_QUERY_TYPE_WRITE;

4db61af2cfe2b206113bcc4b6153521679702bb4Timo Sirainen driver_cassandra_query_full(_ctx->db, _ctx->head->query, query_type,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen transaction_commit_callback, ctx);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen } else {

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen /* multiple queries - we don't actually have a transaction though */

ce74395e2a932342e04fb682395bcce111574969Timo Sirainen callback("Multiple changes in transaction not supported", context);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen }

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainenstatic void

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainencommit_multi_fail(struct cassandra_transaction_context *ctx,

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result *result, const char *query)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen transaction_set_failed(ctx, t_strdup_printf(

e22ec7998afd426c53c658483ce66b6e404e27c6Timo Sirainen "%s (query: %s)", sql_result_get_error(result), query));

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen sql_result_unref(result);

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen}

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainenstatic int

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainendriver_cassandra_transaction_commit_multi(struct cassandra_transaction_context *ctx,

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainen struct sql_result **result_r)

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen{

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct cassandra_db *db = (struct cassandra_db *)ctx->ctx.db;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_result *result;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen struct sql_transaction_query *query;

b650f04c3b2e7dea2295bdbe3239eb82ec03ada0Timo Sirainen int ret = 0;

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen result = driver_cassandra_sync_query(db, "BEGIN");

b772ddf3cfb606dddaa465b317a0dc01bf06c6e4Timo Sirainen if (sql_result_next_row(result) < 0) {