httpserver-pg.h

/* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
*
* MIT License
* 
* Copyright (c) 2020 Jeremy Williams
* 
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
* 
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
* 
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *
*
* httpserver-pg.h (0.1.0)
*
* Description:
*
*   Helper for adding non-blocking PostgreSQL connectivity to your httpserver.h
*   based application. It does, however, not depend on httpserver.h at all and
*   could be potentially used for other applications.
*
* Usage:
*
*   If you include this header in your project you will need to install libpq
*   and link it with -lpq
*
*   Do this:
*      #define HTTPSERVER_PG_IMPL
*   before you include this file in *one* C or C++ file to create the
*   implementation.
*
*   // i.e. it should look like this:
*   #include ...
*   #include ...
*   #include ...
*   #define HTTPSERVER_PG_IMPL
*   #include "httpserver-pg.h"
*
*   There are some #defines that can be configured. This must be done in the
*   same file that you define HTTPSERVER_PG_IMPL These defines have default
*   values and will need to be #undef'd and redefined to configure them.
*
*     HS_DB_CONNECTION_POOL_SIZE - default 25 - The amount of connections to
*       opened with the server on initialization. Even though communication
*       over each connection is non-blocking queries cannot be multiplexed over
*       one connection. Therefore this number amounts to the maximum number of
*       inflight queries to the database that can happen concurrently.
*
*   For more details see the documentation of the interface and the example
*   below.
*
* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */

#ifndef HTTPSERVER_PG_H
#define HTTPSERVER_PG_H

#include <libpq-fe.h>

#ifdef __cplusplus
extern "C" {
#endif

struct hs_db_s;
struct hs_db_conn_s;

// Param structure for passing into hs_db_exec
//   server: the context returned by the hs_db_init call
//   data: for passing in application data or context that will be accessible
//     through hs_db_get_data when the callback executes.
//   callback: function pointer that will get executed when the query results
//     are ready.
struct hs_db_params_s {
  struct hs_db_s* server;
  void* data;
  void (*callback)(struct hs_db_conn_s*);
};

// Initialize the pool of connections to the database. connect_str is the string
// that will be passed into libpq. Details can be seen here:
//   https://www.postgresql.org/docs/12/libpq-connect.html#LIBPQ-CONNSTRING
// loop is the file descriptor of the epoll or kqueue event queue.
struct hs_db_s* hs_db_init(char const* connect_str, int loop);

// Execute the passed in query in the context of the params.
int hs_db_exec(struct hs_db_params_s params, char const* query);

// Just a utility method to easily create the hs_db_params_s struct based on the
// given arguments.
struct hs_db_params_s hs_db_exec_params(
  struct hs_db_s* server,
  void* data,
  void (*cb)(struct hs_db_conn_s*)
);

// Just another way to execute a query if you don't want to bundle the params
// into a hs_db_params_s struct. See the description of hs_db_params_s to
// understand what the parameters are.
int hs_db_execp(
  struct hs_db_s* server,
  char const * query,
  void* data,
  void (*cb)(struct hs_db_conn_s*)
);

// For use inside the callback to retrieve the results of the query. See
// https://www.postgresql.org/docs/9.5/libpq-exec.html#LIBPQ-EXEC-SELECT-INFO
// for details on how to work with the result.
PGresult* hs_db_get_result(struct hs_db_conn_s* conn);

// For use inside the callback to retrieve the application data that was passed
// into hs_db_exec via the data pointer.
void* hs_db_get_data(struct hs_db_conn_s* conn);

#ifdef __cplusplus
}
#endif

// *** Example Program ***

#ifdef HS_PG_EXAMPLE

#define HTTPSERVER_IMPL
#include "httpserver.h"
#define HTTPSERVER_PG_IMPL
#include "httpserver-pg.h"

struct hs_db_s* db;

void result_cb(struct hs_db_conn_s* conn) {
  struct http_response_s* response = http_response_init();
  struct http_request_s* request = (struct http_request_s*)hs_db_get_data(conn);
  PGresult* result = hs_db_get_result(conn);
  char* name = PQfname(result, 0);
  http_response_status(response, 200);
  http_response_body(response, name, strlen(name));
  http_respond(request, response);
}

void handle_request(struct http_request_s* request) {
  hs_db_execp(db, "select 1 as test_col", request, result_cb);
}

int main() {
  struct http_server_s* server = http_server_init(8080, handle_request);
  db = hs_db_init("postgres://localhost:5432/jeremy", http_server_loop(server));
  http_server_listen(server);
}

#endif //HS_PG_EXAMPLE

#ifdef HTTPSERVER_PG_IMPL

#ifdef __linux__
#define HS_PG_EPOLL
#else
#define HS_PG_KQUEUE
#endif

#define HS_DB_CONNECTION_POOL_SIZE 25

typedef struct hs_db_conn_s {
#ifdef HS_PG_KQUEUE
  void (*handler)(struct kevent*);
#else
  void (*handler)(struct epoll_event*);
#endif
  PGconn* dbconn;
  PGresult* result;
  void* data;
  struct hs_db_s* server;
  void (*result_cb)(struct hs_db_conn_s* conn);
  struct hs_db_conn_s* next;
} hs_db_conn_t;

typedef struct hs_db_s {
  int loop;
  hs_db_conn_t* avail_conn_head;
  hs_db_conn_t pool[HS_DB_CONNECTION_POOL_SIZE];
} hs_db_t;

#ifdef HS_PG_KQUEUE
void hs_db_read_cb(struct kevent* ev);
#else
void hs_db_read_cb(struct epoll_event* ev);
#endif

void hs_db_add_read_events(hs_db_conn_t* conn);

typedef struct hs_db_params_s hs_db_params_t;

void hs_db_add_avail_conn(hs_db_t* db, hs_db_conn_t* conn) {
  hs_db_conn_t* next = db->avail_conn_head;
  conn->next = next;
  db->avail_conn_head = conn;
}

hs_db_t* hs_db_init(char const * connect_str, int loop) {
  hs_db_t* db = (hs_db_t*)malloc(sizeof(hs_db_t));
  db->loop = loop;
  for (int i = 0; i < HS_DB_CONNECTION_POOL_SIZE; i++) {
    PGconn* dbconn = PQconnectdb(connect_str);
    PQsetnonblocking(dbconn, 1);
    hs_db_conn_t* conn = &db->pool[i];
    conn->dbconn = dbconn;
    conn->handler = hs_db_read_cb;
    conn->server = db;
    hs_db_add_avail_conn(db, conn);
    hs_db_add_read_events(conn);
  }
  return db;
}

hs_db_params_t hs_db_exec_params(hs_db_t* server, void* data, void (*cb)(hs_db_conn_t*)) {
  return (hs_db_params_t) {
    .server = server,
    .data = data,
    .callback = cb
  };
}

int hs_db_exec(hs_db_params_t params, char const* query) {
  if (params.server->avail_conn_head == NULL) return 0;
  hs_db_conn_t* conn = params.server->avail_conn_head;
  params.server->avail_conn_head = conn->next;
  conn->result_cb = params.callback;
  conn->data = params.data;
  return PQsendQuery(conn->dbconn, query);
}

int hs_db_execp(hs_db_t* server, char const * query, void* data, void (*cb)(hs_db_conn_t*)) {
  hs_db_params_t params = hs_db_exec_params(server, data, cb);
  return hs_db_exec(params, query);
}

void hs_db_read_conn(hs_db_conn_t* conn) {
  PQconsumeInput(conn->dbconn);
  PGresult* res;
  while (!PQisBusy(conn->dbconn) && (res = PQgetResult(conn->dbconn))) {
    conn->result = res;
    conn->result_cb(conn);
    PQclear(res);
  }
  if (res == NULL) {
    hs_db_add_avail_conn(conn->server, conn);
  }
}

PGresult* hs_db_get_result(hs_db_conn_t* conn) {
  return conn->result;
}

void* hs_db_get_data(hs_db_conn_t* conn) {
  return conn->data;
}

#ifdef HS_PG_KQUEUE

void hs_db_read_cb(struct kevent* ev) {
  hs_db_read_conn((hs_db_conn_t*)ev->udata);
}

void hs_db_add_read_events(hs_db_conn_t* conn) {
  struct kevent ev_set;
  EV_SET(&ev_set, PQsocket(conn->dbconn), EVFILT_READ, EV_ADD, 0, 0, conn);
  kevent(conn->server->loop, &ev_set, 1, NULL, 0, NULL);
}

#else

void hs_db_add_read_events(hs_db_conn_t* conn) {
  struct epoll_event ev;
  ev.events = EPOLLIN | EPOLLET;
  ev.data.ptr = conn;
  epoll_ctl(conn->server->loop, EPOLL_CTL_ADD, PQsocket(conn->dbconn), &ev);
}

void hs_db_read_cb(struct epoll_event* ev) {
  hs_db_read_conn((hs_db_conn_t*)ev->data.ptr);
}

#endif //HS_PG_KQUEUE

#endif //HTTPSERVER_PG_IMPL

#endif //HTTPSERVER_PG_H