1 

     2 #include <event2/event.h>

     3 #include <assert.h>

     4 #include <stdlib.h>

     5 

     6 #include "evpq.h"

     7 #include "lib/error.h"

     8 

     9 struct evpq_conn {

    10     struct event_base *ev_base;

    11     struct evpq_callback_info user_cb;

    12 

    13     PGconn *pg_conn;

    14 

    15     struct event *ev;

    16 

    17     enum evpq_state state;

    18 };

    19 

    20 /*

    21  * This evpq_conn has experienced a GENERAL FAILURE.

    22  */

    23 static void _evpq_failure (struct evpq_conn *conn) {

    24     // update state

    25     conn->state = EVPQ_FAILURE;

    26 

    27     // notify

    28     conn->user_cb.fn_failure(conn, conn->user_cb.cb_arg);

    29 }

    30 

    31 /*

    32  * Initial connect was succesfull

    33  */

    34 static void _evpq_connect_ok (struct evpq_conn *conn) {

    35     // update state

    36     conn->state = EVPQ_CONNECTED;

    37 

    38     // notify

    39     conn->user_cb.fn_connected(conn, conn->user_cb.cb_arg);

    40 }

    41 

    42 /*

    43  * Initial connect failed

    44  */

    45 static void _evpq_connect_fail (struct evpq_conn *conn) {

    46     // just mark it as a generic failure

    47     _evpq_failure(conn);

    48 }

    49 

    50 /*

    51  * Receive a result and gives it to the user. If there was no more results, update state and tell the user.

    52  *

    53  * Returns zero if we got a result, 1 if there were/are no more results to handle.

    54  */

    55 static int _evpq_query_result (struct evpq_conn *conn) {

    56     PGresult *result;

    57     

    58     // get the result

    59     if ((result = PQgetResult(conn->pg_conn)) == NULL) {

    60         // no more results, update state

    61         conn->state = EVPQ_CONNECTED;

    62 

    63         // tell the user the query is done

    64         conn->user_cb.fn_done(conn, conn->user_cb.cb_arg);

    65 

    66         // stop waiting for more results

    67         return 1;

    68 

    69     } else {

    70         // got a result, give it to the user

    71         conn->user_cb.fn_result(conn, result, conn->user_cb.cb_arg);

    72 

    73         // great

    74         return 0;

    75     }

    76 }

    77 

    78 /*

    79  * Schedule a new _evpq_event for this connection.

    80  */ 

    81 static int _evpq_schedule (struct evpq_conn *conn, short what, void (*handler)(evutil_socket_t, short, void *)) {

    82     assert(conn->pg_conn != NULL);

    83 

    84     // ensure we have a valid socket, this should be the case after the PQstatus check...

    85     if (PQsocket(conn->pg_conn) < 0)

    86         FATAL("PQsocket gave invalid socket");

    87 

    88     // reschedule with a new event

    89     if (conn->ev) {

    90         event_assign(conn->ev, conn->ev_base, PQsocket(conn->pg_conn), what, handler, conn);

    91 

    92     } else {

    93         if ((conn->ev = event_new(conn->ev_base, PQsocket(conn->pg_conn), what, handler, conn)) == NULL)

    94             PERROR("event_new");

    95 

    96     }

    97 

    98     // add it

    99     // XXX: timeouts?

   100     if (event_add(conn->ev, NULL))

   101         PERROR("event_add");

   102     

   103     // success

   104     return 0;

   105 

   106 error:

   107     return -1;

   108 }

   109 

   110 /*

   111  * Handle events on the PQ socket while connecting

   112  */

   113 static void _evpq_connect_event (evutil_socket_t fd, short what, void *arg) {

   114     struct evpq_conn *conn = arg;

   115     PostgresPollingStatusType poll_status;

   116     

   117     // this is only for connect events

   118     assert(conn->state == EVPQ_CONNECT);

   119     

   120     // XXX: timeouts?

   121 

   122     // ask PQ what to do 

   123     switch ((poll_status = PQconnectPoll(conn->pg_conn))) {

   124         case PGRES_POLLING_READING:

   125             // poll for read

   126             what = EV_READ;

   127             

   128             // reschedule

   129             break;

   130 

   131         case PGRES_POLLING_WRITING:

   132             // poll for write

   133             what = EV_WRITE;

   134             

   135             // reschedule

   136             break;

   137 

   138         case PGRES_POLLING_OK:

   139             // connected

   140             _evpq_connect_ok(conn);

   141             

   142             // done

   143             return;

   144         

   145         case PGRES_POLLING_FAILED:

   146             // faaaaail!

   147             _evpq_connect_fail(conn);

   148 

   149             // done

   150             return;

   151         

   152         default:

   153             FATAL("PQconnectPoll gave a weird value: %d", poll_status);

   154     }

   155     

   156     // reschedule

   157     if (_evpq_schedule(conn, what, _evpq_connect_event))

   158         goto error;

   159 

   160     // done, wait for the next event

   161     return;

   162 

   163 error:

   164     // XXX: reset?

   165     _evpq_failure(conn);

   166 }

   167 

   168 static void _evpq_query_event (evutil_socket_t fd, short what, void *arg) {

   169     struct evpq_conn *conn = arg;

   170     

   171     // this is only for query events

   172     assert(conn->state == EVPQ_QUERY);

   173 

   174     // XXX: PQflush, timeouts

   175     assert(what == EV_READ);

   176 

   177     // we're going to assume that all queries will *require* data for their results

   178     // this would break otherwise (PQconsumeInput might block?)

   179     assert(PQisBusy(conn->pg_conn) != 0);

   180 

   181     // handle input

   182     if (PQconsumeInput(conn->pg_conn) == 0)

   183         ERROR("PQconsumeInput: %s", PQerrorMessage(conn->pg_conn));

   184     

   185     // handle results

   186     while (PQisBusy(conn->pg_conn) == 0) {

   187         // handle the result

   188         if (_evpq_query_result(conn) == 1) {

   189             // no need to wait for anything anymore

   190             return;

   191         }

   192 

   193         // loop to handle the next result

   194     }

   195 

   196     // still need to wait for a result, so reschedule

   197     if (_evpq_schedule(conn, EV_READ, _evpq_query_event))

   198         goto error;

   199         

   200     // done, wait for the next event

   201     return;

   202 

   203 error:

   204     // XXX: reset?

   205     _evpq_failure(conn);

   206 

   207 }

   208 

   209 struct evpq_conn *evpq_connect (struct event_base *ev_base, const char *conninfo, const struct evpq_callback_info cb_info) {

   210     struct evpq_conn *conn = NULL;

   211     

   212     // alloc our context

   213     if ((conn = calloc(1, sizeof(*conn))) == NULL)

   214         ERROR("calloc");

   215     

   216     // initial state

   217     conn->ev_base = ev_base;

   218     conn->user_cb = cb_info;

   219     conn->state = EVPQ_INIT;

   220 

   221     // create our PGconn

   222     if ((conn->pg_conn = PQconnectStart(conninfo)) == NULL)

   223         PERROR("PQconnectStart");

   224     

   225     // check for immediate failure

   226     if (PQstatus(conn->pg_conn) == CONNECTION_BAD)

   227         ERROR("PQstatus indicates CONNECTION_BAD after PQconnectStart");

   228     

   229     // assume PGRES_POLLING_WRITING

   230     if (_evpq_schedule(conn, EV_WRITE, _evpq_connect_event))

   231         goto error;

   232 

   233     // connecting

   234     conn->state = EVPQ_CONNECT;

   235 

   236     // success, wait for the connection to be established

   237     return conn;

   238 

   239 error:

   240     if (conn) {

   241         if (conn->pg_conn)

   242             PQfinish(conn->pg_conn);

   243         

   244         free(conn);

   245     }

   246 

   247     return NULL;

   248 }

   249 

   250 int evpq_query (struct evpq_conn *conn, const char *command) {

   251     // check state

   252     if (conn->state != EVPQ_CONNECTED)

   253         ERROR("invalid evpq state: %d", conn->state);

   254     

   255     // do the query

   256     if (PQsendQuery(conn->pg_conn, command) == 0)

   257         ERROR("PQsendQuery: %s", PQerrorMessage(conn->pg_conn));

   258     

   259     // update state

   260     conn->state = EVPQ_QUERY;

   261     

   262     // XXX: PQflush

   263 

   264     // poll for read

   265     if (_evpq_schedule(conn, EV_READ, _evpq_query_event))

   266         goto error;

   267 

   268     // and then we wait

   269     return 0;

   270 

   271 error:

   272     return -1;

   273 }

   274 

   275 const PGconn *evpq_pgconn (struct evpq_conn *conn) {

   276     return conn->pg_conn;

   277 }