1

#define _GNU_SOURCE

     2

#include <stdlib.h>

     3

#include <sys/queue.h>

     4

#include <assert.h>

     5

#include <string.h>

     6

     7

#include "evsql.h"

     8

#include "evpq.h"

     9

#include "lib/log.h"

    10

#include "lib/error.h"

    11

#include "lib/misc.h"

    12

    13

enum evsql_type {

    14

    EVSQL_EVPQ,

    15

};

    16

    17

struct evsql {

    18

    // callbacks

    19

    evsql_error_cb error_fn;

    20

    void *cb_arg;

    21

    22

    // backend engine

    23

    enum evsql_type type;

    24

    25

    union {

    26

        struct evpq_conn *evpq;

    27

    } engine;

    28

    29

    // list of queries running or waiting to run

    30

    TAILQ_HEAD(evsql_queue, evsql_query) queue;

    31

};

    32

    33

struct evsql_query {

    34

    // the evsql we are querying

    35

    struct evsql *evsql;

    36

    37

    // the actual SQL query, this may or may not be ours, see _evsql_query_exec

    38

    char *command;

    39

    40

    // our callback

    41

    evsql_query_cb cb_fn;

    42

    void *cb_arg;

    43

    44

    // our position in the query list

    45

    TAILQ_ENTRY(evsql_query) entry;

    46

    47

    // the result

    48

    union {

    49

        PGresult *evpq;

    50

    } result;

    51

};

    52

    53

/*

    54

 * Actually execute the given query.

    55

 *

    56

 * The backend should be able to accept the query at this time.

    57

 *

    58

 * query->command must be valid during the execution of this function, but once it returns, the command is not needed

    59

 * anymore, and should be set to NULL.

    60

 */

    61

static int _evsql_query_exec (struct evsql *evsql, struct evsql_query *query, const char *command) {

    62

    switch (evsql->type) {

    63

        case EVSQL_EVPQ:

    64

            // just pass it through

    65

            return evpq_query(evsql->engine.evpq, command);

    66

    67

        default:

    68

            FATAL("evsql->type");

    69

    }

    70

}

    71

    72

/*

    73

 * Dequeue the query, execute the callback, and free it.

    74

 */

    75

static void _evsql_query_done (struct evsql_query *query, const struct evsql_result_info *result_info) {

    76

    assert(query->command == NULL);

    77

    78

    // dequeue

    79

    TAILQ_REMOVE(&query->evsql->queue, query, entry);

    80

    81

    if (result_info) 

    82

        // call the callback

    83

        query->cb_fn(*result_info, query->cb_arg);

    84

    85

    // free

    86

    free(query);

    87

}

    88

    89

/*

    90

 * A query has failed, notify the user and remove it.

    91

 */

    92

static void _evsql_query_failure (struct evsql *evsql, struct evsql_query *query) {

    93

    struct evsql_result_info result; ZINIT(result);

    94

    95

    // set up the result_info

    96

    result.evsql = evsql;

    97

    result.error = 1;

    98

    99

    // finish it off

   100

    _evsql_query_done(query, &result);

   101

}

   102

   103

/*

   104

 * Clear every enqueued query and then free the evsql.

   105

 *

   106

 * If result_info is given, each query will also recieve it via their callback, and the error_fn will be called.

   107

 */

   108

static void _evsql_destroy (struct evsql *evsql, const struct evsql_result_info *result_info) {

   109

    struct evsql_query *query;

   110

   111

    // clear the queue

   112

    while ((query = TAILQ_FIRST(&evsql->queue)) != NULL) {

   113

        _evsql_query_done(query, result_info);

   114

   115

        TAILQ_REMOVE(&evsql->queue, query, entry);

   116

    }

   117

   118

    // do the error callback if required

   119

    if (result_info)

   120

        evsql->error_fn(evsql, evsql->cb_arg);

   121

   122

    // free

   123

    free(evsql);

   124

}

   125

   126

   127

/*

   128

 * Sends the next query if there are more enqueued

   129

 */

   130

static void _evsql_pump (struct evsql *evsql) {

   131

    struct evsql_query *query;

   132

   133

    // look for the next query

   134

    if ((query = TAILQ_FIRST(&evsql->queue)) != NULL) {

   135

        // try and execute it

   136

        if (_evsql_query_exec(evsql, query, query->command)) {

   137

            // the query failed

   138

            _evsql_query_failure(evsql, query);

   139

        }

   140

   141

        // free the command

   142

        free(query->command); query->command = NULL;

   143

   144

        // ok, then we just wait

   145

    }

   146

}

   147

   148

   149

static void _evsql_evpq_connected (struct evpq_conn *conn, void *arg) {

   150

    struct evsql *evsql = arg;

   151

   152

    // no state to update, just pump any waiting queries

   153

    _evsql_pump(evsql);

   154

}

   155

   156

static void _evsql_evpq_result (struct evpq_conn *conn, PGresult *result, void *arg) {

   157

    struct evsql *evsql = arg;

   158

    struct evsql_query *query;

   159

   160

    assert((query = TAILQ_FIRST(&evsql->queue)) != NULL);

   161

   162

    // if we get multiple results, only return the first one

   163

    if (query->result.evpq) {

   164

        WARNING("[evsql] evpq query returned multiple results, discarding previous one");

   165

   166

        PQclear(query->result.evpq); query->result.evpq = NULL;

   167

    }

   168

   169

    // remember the result

   170

    query->result.evpq = result;

   171

}

   172

   173

static void _evsql_evpq_done (struct evpq_conn *conn, void *arg) {

   174

    struct evsql *evsql = arg;

   175

    struct evsql_query *query;

   176

    struct evsql_result_info result; ZINIT(result);

   177

   178

    assert((query = TAILQ_FIRST(&evsql->queue)) != NULL);

   179

   180

    // set up the result_info

   181

    result.evsql = evsql;

   182

   183

    if (query->result.evpq == NULL) {

   184

        // if a query didn't return any results (bug?), warn and fail the query

   185

        WARNING("[evsql] evpq query didn't return any results");

   186

   187

        result.error = 1;

   188

   189

    } else {

   190

        result.error = 0;

   191

        result.result.pq = query->result.evpq;

   192

   193

    }

   194

   195

    // finish it off

   196

    _evsql_query_done(query, &result);

   197

   198

    // pump the next one

   199

    _evsql_pump(evsql);

   200

}

   201

   202

static void _evsql_evpq_failure (struct evpq_conn *conn, void *arg) {

   203

    struct evsql *evsql = arg;

   204

    struct evsql_result_info result; ZINIT(result);

   205

   206

    // OH SHI...

   207

   208

    // set up the result_info

   209

    result.evsql = evsql;

   210

    result.error = 1;

   211

   212

    // finish off the whole connection

   213

    _evsql_destroy(evsql, &result);

   214

}

   215

   216

static struct evpq_callback_info _evsql_evpq_cb_info = {

   217

    .fn_connected       = _evsql_evpq_connected,

   218

    .fn_result          = _evsql_evpq_result,

   219

    .fn_done            = _evsql_evpq_done,

   220

    .fn_failure         = _evsql_evpq_failure,

   221

};

   222

   223

static struct evsql *_evsql_new_base (evsql_error_cb error_fn, void *cb_arg) {

   224

    struct evsql *evsql = NULL;

   225

   226

    // allocate it

   227

    if ((evsql = calloc(1, sizeof(*evsql))) == NULL)

   228

        ERROR("calloc");

   229

   230

    // store

   231

    evsql->error_fn = error_fn;

   232

    evsql->cb_arg = cb_arg;

   233

   234

    // init

   235

    TAILQ_INIT(&evsql->queue);

   236

   237

    // done

   238

    return evsql;

   239

   240

error:

   241

    return NULL;

   242

}

   243

   244

struct evsql *evsql_new_pq (struct event_base *ev_base, const char *pq_conninfo, evsql_error_cb error_fn, void *cb_arg) {

   245

    struct evsql *evsql = NULL;

   246

   247

    // base init

   248

    if ((evsql = _evsql_new_base (error_fn, cb_arg)) == NULL)

   249

        goto error;

   250

   251

    // connect the engine

   252

    if ((evsql->engine.evpq = evpq_connect(ev_base, pq_conninfo, _evsql_evpq_cb_info, evsql)) == NULL)

   253

        goto error;

   254

   255

    // done

   256

    return evsql;

   257

   258

error:

   259

    // XXX: more complicated than this?

   260

    free(evsql); 

   261

   262

    return NULL;

   263

}

   264

   265

/*

   266

 * Checks what the state of the connection is in regards to executing a query.

   267

 *

   268

 * Returns:

   269

 *      <0      connection failure, query not possible

   270

 *      0       connection idle, can query immediately

   271

 *      1       connection busy, must queue query

   272

 */

   273

static int _evsql_query_idle (struct evsql *evsql) {

   274

    switch (evsql->type) {

   275

        case EVSQL_EVPQ: {

   276

            enum evpq_state state = evpq_state(evsql->engine.evpq);

   277

   278

            switch (state) {

   279

                case EVPQ_CONNECT:

   280

                case EVPQ_QUERY:

   281

                    return 1;

   282

   283

                case EVPQ_CONNECTED:

   284

                    return 0;

   285

   286

                case EVPQ_INIT:

   287

                case EVPQ_FAILURE:

   288

                    return -1;

   289

   290

                default:

   291

                    FATAL("evpq_state");

   292

            }

   293

   294

        }

   295

   296

        default:

   297

            FATAL("evsql->type");

   298

    }

   299

}

   300

   301

   302

struct evsql_query *evsql_query (struct evsql *evsql, const char *command, evsql_query_cb query_fn, void *cb_arg) {

   303

    struct evsql_query *query;

   304

    int idle;

   305

   306

    // allocate it

   307

    if ((query = calloc(1, sizeof(*query))) == NULL)

   308

        ERROR("calloc");

   309

   310

    // store

   311

    query->evsql = evsql;

   312

    query->cb_fn = query_fn;

   313

    query->cb_arg = cb_arg;

   314

   315

    // check state

   316

    if ((idle = _evsql_query_idle(evsql)) < 0)

   317

        ERROR("connection is not valid");

   318

   319

    if (idle) {

   320

        assert(TAILQ_EMPTY(&evsql->queue));

   321

   322

        // execute directly

   323

        if (_evsql_query_exec(evsql, query, command))

   324

            goto error;

   325

   326

    } else {

   327

        // copy the command for later execution

   328

        if ((query->command = strdup(command)) == NULL)

   329

            ERROR("strdup");

   330

    }

   331

   332

    // store it on the list

   333

    TAILQ_INSERT_TAIL(&evsql->queue, query, entry);

   334

   335

    // success

   336

    return query;

   337

   338

error:

   339

    // do *NOT* free query->command, ever

   340

    free(query);

   341

   342

    return NULL;

   343

}

   344