/* connection_int.c - code used by the connection object
 *
 * Copyright (C) 2003-2019 Federico Di Gregorio <fog@debian.org>
 *
 * This file is part of psycopg.
 *
 * psycopg2 is free software: you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published
 * by the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * In addition, as a special exception, the copyright holders give
 * permission to link this program with the OpenSSL library (or with
 * modified versions of OpenSSL that use the same license as OpenSSL),
 * and distribute linked combinations including the two.
 *
 * You must obey the GNU Lesser General Public License in all respects for
 * all of the code used other than OpenSSL.
 *
 * psycopg2 is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public
 * License for more details.
 */

#define PSYCOPG_MODULE
#include "psycopg/psycopg.h"

#include "psycopg/connection.h"
#include "psycopg/cursor.h"
#include "psycopg/pqpath.h"
#include "psycopg/green.h"
#include "psycopg/notify.h"

#include <string.h>

/* String indexes match the ISOLATION_LEVEL_* consts */
const char *srv_isolevels[] = {
    NULL, /* autocommit */
    "READ COMMITTED",
    "REPEATABLE READ",
    "SERIALIZABLE",
    "READ UNCOMMITTED",
    "default"       /* only to set GUC, not for BEGIN */
};

/* Read only false, true */
const char *srv_readonly[] = {
    " READ WRITE",
    " READ ONLY",
    ""      /* default */
};

/* Deferrable false, true */
const char *srv_deferrable[] = {
    " NOT DEFERRABLE",
    " DEFERRABLE",
    ""      /* default */
};

/* On/Off/Default GUC states
 */
const char *srv_state_guc[] = {
    "off",
    "on",
    "default"
};


const int SRV_STATE_UNCHANGED = -1;


/* Return a new "string" from a char* from the database.
 *
 * On Py2 just get a string, on Py3 decode it in the connection codec.
 *
 * Use a fallback if the connection is NULL.
 */
PyObject *
conn_text_from_chars(connectionObject *self, const char *str)
{
    return psyco_text_from_chars_safe(str, -1, self ? self->pydecoder : NULL);
}


/* Encode an unicode object into a bytes object in the connection encoding.
 *
 * If no connection or encoding is available, default to utf8
 */
PyObject *
conn_encode(connectionObject *self, PyObject *u)
{
    PyObject *t = NULL;
    PyObject *rv = NULL;

    if (!(self && self->pyencoder)) {
        rv = PyUnicode_AsUTF8String(u);
        goto exit;
    }

    if (!(t = PyObject_CallFunctionObjArgs(self->pyencoder, u, NULL))) {
        goto exit;
    }

    if (!(rv = PyTuple_GetItem(t, 0))) { goto exit; }
    Py_INCREF(rv);

exit:
    Py_XDECREF(t);

    return rv;
}


/* decode a c string into a Python unicode in the connection encoding
 *
 * len can be < 0: in this case it will be calculated
 *
 * If no connection or encoding is available, default to utf8
 */
PyObject *
conn_decode(connectionObject *self, const char *str, Py_ssize_t len)
{
    if (len < 0) { len = strlen(str); }

    if (self) {
        if (self->cdecoder) {
            return self->cdecoder(str, len, NULL);
        }
        else if (self->pydecoder) {
            PyObject *b = NULL;
            PyObject *t = NULL;
            PyObject *rv = NULL;

            if (!(b = Bytes_FromStringAndSize(str, len))) { goto error; }
            if (!(t = PyObject_CallFunctionObjArgs(self->pydecoder, b, NULL))) {
                goto error;
            }
            if (!(rv = PyTuple_GetItem(t, 0))) { goto error; }
            Py_INCREF(rv);      /* PyTuple_GetItem gives a borrowes one */
error:
            Py_XDECREF(t);
            Py_XDECREF(b);
            return rv;
        }
        else {
            return PyUnicode_FromStringAndSize(str, len);
        }
    }
    else {
        return PyUnicode_FromStringAndSize(str, len);
    }
}

/* conn_notice_callback - process notices */

static void
conn_notice_callback(void *args, const char *message)
{
    struct connectionObject_notice *notice;
    connectionObject *self = (connectionObject *)args;

    Dprintf("conn_notice_callback: %s", message);

    /* NOTE: if we get here and the connection is unlocked then there is a
       problem but this should happen because the notice callback is only
       called from libpq and when we're inside libpq the connection is usually
       locked.
    */
    notice = (struct connectionObject_notice *)
        malloc(sizeof(struct connectionObject_notice));
    if (NULL == notice) {
        /* Discard the notice in case of failed allocation. */
        return;
    }
    notice->next = NULL;
    notice->message = strdup(message);
    if (NULL == notice->message) {
        free(notice);
        return;
    }

    if (NULL == self->last_notice) {
        self->notice_pending = self->last_notice = notice;
    }
    else {
        self->last_notice->next = notice;
        self->last_notice = notice;
    }
}

/* Expose the notices received as Python objects.
 *
 * The function should be called with the connection lock and the GIL.
 */
void
conn_notice_process(connectionObject *self)
{
    struct connectionObject_notice *notice;
    PyObject *msg = NULL;
    PyObject *tmp = NULL;
    static PyObject *append;

    if (NULL == self->notice_pending) {
        return;
    }

    if (!append) {
        if (!(append = Text_FromUTF8("append"))) {
            goto error;
        }
    }

    notice = self->notice_pending;
    while (notice != NULL) {
        Dprintf("conn_notice_process: %s", notice->message);

        if (!(msg = conn_text_from_chars(self, notice->message))) { goto error; }

        if (!(tmp = PyObject_CallMethodObjArgs(
                self->notice_list, append, msg, NULL))) {

            goto error;
        }

        Py_DECREF(tmp); tmp = NULL;
        Py_DECREF(msg); msg = NULL;

        notice = notice->next;
    }

    /* Remove the oldest item if the queue is getting too long. */
    if (PyList_Check(self->notice_list)) {
        Py_ssize_t nnotices;
        nnotices = PyList_GET_SIZE(self->notice_list);
        if (nnotices > CONN_NOTICES_LIMIT) {
            if (-1 == PySequence_DelSlice(self->notice_list,
                    0, nnotices - CONN_NOTICES_LIMIT)) {
                PyErr_Clear();
            }
        }
    }

    conn_notice_clean(self);
    return;

error:
    Py_XDECREF(tmp);
    Py_XDECREF(msg);
    conn_notice_clean(self);

    /* TODO: the caller doesn't expects errors from us */
    PyErr_Clear();
}

void
conn_notice_clean(connectionObject *self)
{
    struct connectionObject_notice *tmp, *notice;

    notice = self->notice_pending;

    while (notice != NULL) {
        tmp = notice;
        notice = notice->next;
        free(tmp->message);
        free(tmp);
    }

    self->last_notice = self->notice_pending = NULL;
}


/* conn_notifies_process - make received notification available
 *
 * The function should be called with the connection lock and holding the GIL.
 */

void
conn_notifies_process(connectionObject *self)
{
    PGnotify *pgn = NULL;
    PyObject *notify = NULL;
    PyObject *pid = NULL, *channel = NULL, *payload = NULL;
    PyObject *tmp = NULL;

    static PyObject *append;

    if (!append) {
        if (!(append = Text_FromUTF8("append"))) {
            goto error;
        }
    }

    while ((pgn = PQnotifies(self->pgconn)) != NULL) {

        Dprintf("conn_notifies_process: got NOTIFY from pid %d, msg = %s",
                (int) pgn->be_pid, pgn->relname);

        if (!(pid = PyInt_FromLong((long)pgn->be_pid))) { goto error; }
        if (!(channel = conn_text_from_chars(self, pgn->relname))) { goto error; }
        if (!(payload = conn_text_from_chars(self, pgn->extra))) { goto error; }

        if (!(notify = PyObject_CallFunctionObjArgs((PyObject *)&notifyType,
                pid, channel, payload, NULL))) {
            goto error;
        }

        Py_DECREF(pid); pid = NULL;
        Py_DECREF(channel); channel = NULL;
        Py_DECREF(payload); payload = NULL;

        if (!(tmp = PyObject_CallMethodObjArgs(
                self->notifies, append, notify, NULL))) {
            goto error;
        }
        Py_DECREF(tmp); tmp = NULL;

        Py_DECREF(notify); notify = NULL;
        PQfreemem(pgn); pgn = NULL;
    }
    return;  /* no error */

error:
    if (pgn) { PQfreemem(pgn); }
    Py_XDECREF(tmp);
    Py_XDECREF(notify);
    Py_XDECREF(pid);
    Py_XDECREF(channel);
    Py_XDECREF(payload);

    /* TODO: callers currently don't expect an error from us */
    PyErr_Clear();

}


/*
 * the conn_get_* family of functions makes it easier to obtain the connection
 * parameters from query results or by interrogating the connection itself
*/

int
conn_get_standard_conforming_strings(PGconn *pgconn)
{
    int equote;
    const char *scs;
    /*
     * The presence of the 'standard_conforming_strings' parameter
     * means that the server _accepts_ the E'' quote.
     *
     * If the parameter is off, the PQescapeByteaConn returns
     * backslash escaped strings (e.g. '\001' -> "\\001"),
     * so the E'' quotes are required to avoid warnings
     * if 'escape_string_warning' is set.
     *
     * If the parameter is on, the PQescapeByteaConn returns
     * not escaped strings (e.g. '\001' -> "\001"), relying on the
     * fact that the '\' will pass untouched the string parser.
     * In this case the E'' quotes are NOT to be used.
     */
    scs = PQparameterStatus(pgconn, "standard_conforming_strings");
    Dprintf("conn_connect: server standard_conforming_strings parameter: %s",
        scs ? scs : "unavailable");

    equote = (scs && (0 == strcmp("off", scs)));
    Dprintf("conn_connect: server requires E'' quotes: %s",
            equote ? "YES" : "NO");

    return equote;
}


/* Remove irrelevant chars from encoding name and turn it uppercase.
 *
 * Return a buffer allocated on Python heap into 'clean' and return 0 on
 * success, otherwise return -1 and set an exception.
 */
RAISES_NEG static int
clear_encoding_name(const char *enc, char **clean)
{
    const char *i = enc;
    char *j, *buf;
    int rv = -1;

    /* convert to upper case and remove '-' and '_' from string */
    if (!(j = buf = PyMem_Malloc(strlen(enc) + 1))) {
        PyErr_NoMemory();
        goto exit;
    }

    while (*i) {
        if (!isalnum(*i)) {
            ++i;
        }
        else {
            *j++ = toupper(*i++);
        }
    }
    *j = '\0';

    Dprintf("clear_encoding_name: %s -> %s", enc, buf);
    *clean = buf;
    rv = 0;

exit:
    return rv;
}

/* set fast access functions according to the currently selected encoding
 */
static void
conn_set_fast_codec(connectionObject *self)
{
    Dprintf("conn_set_fast_codec: encoding=%s", self->encoding);

    if (0 == strcmp(self->encoding, "UTF8")) {
        Dprintf("conn_set_fast_codec: PyUnicode_DecodeUTF8");
        self->cdecoder = PyUnicode_DecodeUTF8;
        return;
    }

    if (0 == strcmp(self->encoding, "LATIN1")) {
        Dprintf("conn_set_fast_codec: PyUnicode_DecodeLatin1");
        self->cdecoder = PyUnicode_DecodeLatin1;
        return;
    }

    Dprintf("conn_set_fast_codec: no fast codec");
    self->cdecoder = NULL;
}


/* Return the Python encoding from a PostgreSQL encoding.
 *
 * Optionally return the clean version of the postgres encoding too
 */
PyObject *
conn_pgenc_to_pyenc(const char *encoding, char **clean_encoding)
{
    char *pgenc = NULL;
    PyObject *rv = NULL;

    if (0 > clear_encoding_name(encoding, &pgenc)) { goto exit; }
    if (!(rv = PyDict_GetItemString(psycoEncodings, pgenc))) {
        PyErr_Format(OperationalError,
            "no Python encoding for PostgreSQL encoding '%s'", pgenc);
        goto exit;
    }
    Py_INCREF(rv);

    if (clean_encoding) {
        *clean_encoding = pgenc;
    }
    else {
        PyMem_Free(pgenc);
    }

exit:
    return rv;
}

/* Convert a Postgres encoding into Python encoding and decoding functions.
 *
 * Set clean_encoding to a clean version of the Postgres encoding name
 * and pyenc and pydec to python codec functions.
 *
 * Return 0 on success, else -1 and set an exception.
 */
RAISES_NEG static int
conn_get_python_codec(const char *encoding,
    char **clean_encoding, PyObject **pyenc, PyObject **pydec)
{
    int rv = -1;
    char *pgenc = NULL;
    PyObject *encname = NULL;
    PyObject *enc_tmp = NULL, *dec_tmp = NULL;

    /* get the Python name of the encoding as a C string */
    if (!(encname = conn_pgenc_to_pyenc(encoding, &pgenc))) { goto exit; }
    if (!(encname = psyco_ensure_bytes(encname))) { goto exit; }

    /* Look up the codec functions */
    if (!(enc_tmp = PyCodec_Encoder(Bytes_AS_STRING(encname)))) { goto exit; }
    if (!(dec_tmp = PyCodec_Decoder(Bytes_AS_STRING(encname)))) { goto exit; }

    /* success */
    *pyenc = enc_tmp; enc_tmp = NULL;
    *pydec = dec_tmp; dec_tmp = NULL;
    *clean_encoding = pgenc; pgenc = NULL;
    rv = 0;

exit:
    Py_XDECREF(enc_tmp);
    Py_XDECREF(dec_tmp);
    Py_XDECREF(encname);
    PyMem_Free(pgenc);

    return rv;
}


/* Store the encoding in the pgconn->encoding field and set the other related
 * encoding fields in the connection structure.
 *
 * Return 0 on success, else -1 and set an exception.
 */
RAISES_NEG static int
conn_store_encoding(connectionObject *self, const char *encoding)
{
    int rv = -1;
    char *pgenc = NULL;
    PyObject *enc_tmp = NULL, *dec_tmp = NULL;

    if (0 > conn_get_python_codec(encoding, &pgenc, &enc_tmp, &dec_tmp)) {
        goto exit;
    }

    /* Good, success: store the encoding/codec in the connection. */
    {
        char *tmp = self->encoding;
        self->encoding = pgenc;
        PyMem_Free(tmp);
        pgenc = NULL;
    }

    Py_CLEAR(self->pyencoder);
    self->pyencoder = enc_tmp;
    enc_tmp = NULL;

    Py_CLEAR(self->pydecoder);
    self->pydecoder = dec_tmp;
    dec_tmp = NULL;

    conn_set_fast_codec(self);

    rv = 0;

exit:
    Py_XDECREF(enc_tmp);
    Py_XDECREF(dec_tmp);
    PyMem_Free(pgenc);
    return rv;
}


/* Read the client encoding from the backend and store it in the connection.
 *
 * Return 0 on success, else -1.
 */
RAISES_NEG static int
conn_read_encoding(connectionObject *self, PGconn *pgconn)
{
    const char *encoding;
    int rv = -1;

    encoding = PQparameterStatus(pgconn, "client_encoding");
    Dprintf("conn_connect: client encoding: %s", encoding ? encoding : "(none)");
    if (!encoding) {
        PyErr_SetString(OperationalError,
            "server didn't return client encoding");
        goto exit;
    }

    if (0 > conn_store_encoding(self, encoding)) {
        goto exit;
    }

    rv = 0;

exit:
    return rv;
}


int
conn_get_protocol_version(PGconn *pgconn)
{
    int ret;
    ret = PQprotocolVersion(pgconn);
    Dprintf("conn_connect: using protocol %d", ret);
    return ret;
}

int
conn_get_server_version(PGconn *pgconn)
{
    return (int)PQserverVersion(pgconn);
}

/* set up the cancel key of the connection.
 * On success return 0, else set an exception and return -1
 */
RAISES_NEG static int
conn_setup_cancel(connectionObject *self, PGconn *pgconn)
{
    if (self->cancel) {
        PQfreeCancel(self->cancel);
    }

    if (!(self->cancel = PQgetCancel(self->pgconn))) {
        PyErr_SetString(OperationalError, "can't get cancellation key");
        return -1;
    }

    return 0;
}

/* Return 1 if the "replication" keyword is set in the DSN, 0 otherwise */
static int
dsn_has_replication(char *pgdsn)
{
    int ret = 0;
    PQconninfoOption *connopts, *ptr;

    connopts = PQconninfoParse(pgdsn, NULL);

    for(ptr = connopts; ptr->keyword != NULL; ptr++) {
      if(strcmp(ptr->keyword, "replication") == 0 && ptr->val != NULL)
        ret = 1;
    }

    PQconninfoFree(connopts);

    return ret;
}


/* Return 1 if the server datestyle allows us to work without problems,
   0 if it needs to be set to something better, e.g. ISO. */
static int
conn_is_datestyle_ok(PGconn *pgconn)
{
    const char *ds;

    ds = PQparameterStatus(pgconn, "DateStyle");
    Dprintf("conn_connect: DateStyle %s", ds);

    /* pgbouncer does not pass on DateStyle */
    if (ds == NULL)
      return 0;

    /* Return true if ds starts with "ISO"
     * e.g. "ISO, DMY" is fine, "German" not. */
    return (ds[0] == 'I' && ds[1] == 'S' && ds[2] == 'O');
}


/* conn_setup - setup and read basic information about the connection */

RAISES_NEG int
conn_setup(connectionObject *self)
{
    int rv = -1;

    self->equote = conn_get_standard_conforming_strings(self->pgconn);
    self->server_version = conn_get_server_version(self->pgconn);
    self->protocol = conn_get_protocol_version(self->pgconn);
    if (3 != self->protocol) {
        PyErr_SetString(InterfaceError, "only protocol 3 supported");
        goto exit;
    }

    if (0 > conn_read_encoding(self, self->pgconn)) {
        goto exit;
    }

    if (0 > conn_setup_cancel(self, self->pgconn)) {
        goto exit;
    }

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);
    Py_BLOCK_THREADS;

    if (!dsn_has_replication(self->dsn) && !conn_is_datestyle_ok(self->pgconn)) {
        int res;
        Py_UNBLOCK_THREADS;
        res = pq_set_guc_locked(self, "datestyle", "ISO", &_save);
        Py_BLOCK_THREADS;
        if (res < 0) {
            pq_complete_error(self);
            goto unlock;
        }
    }

    /* for reset */
    self->autocommit = 0;
    self->isolevel = ISOLATION_LEVEL_DEFAULT;
    self->readonly = STATE_DEFAULT;
    self->deferrable = STATE_DEFAULT;

    /* success */
    rv = 0;

unlock:
    Py_UNBLOCK_THREADS;
    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;

exit:
    return rv;
}

/* conn_connect - execute a connection to the database */

static int
_conn_sync_connect(connectionObject *self)
{
    int green;

    /* store this value to prevent inconsistencies due to a change
     * in the middle of the function. */
    green = psyco_green();
    if (!green) {
        Py_BEGIN_ALLOW_THREADS;
        self->pgconn = PQconnectdb(self->dsn);
        Py_END_ALLOW_THREADS;
        Dprintf("conn_connect: new PG connection at %p", self->pgconn);
    }
    else {
        Py_BEGIN_ALLOW_THREADS;
        self->pgconn = PQconnectStart(self->dsn);
        Py_END_ALLOW_THREADS;
        Dprintf("conn_connect: new green PG connection at %p", self->pgconn);
    }

    if (!self->pgconn)
    {
        Dprintf("conn_connect: PQconnectdb(%s) FAILED", self->dsn);
        PyErr_SetString(OperationalError, "PQconnectdb() failed");
        return -1;
    }
    else if (PQstatus(self->pgconn) == CONNECTION_BAD)
    {
        Dprintf("conn_connect: PQconnectdb(%s) returned BAD", self->dsn);
        PyErr_SetString(OperationalError, PQerrorMessage(self->pgconn));
        return -1;
    }

    PQsetNoticeProcessor(self->pgconn, conn_notice_callback, (void*)self);

    /* if the connection is green, wait to finish connection */
    if (green) {
        if (0 > pq_set_non_blocking(self, 1)) {
            return -1;
        }
        if (0 != psyco_wait(self)) {
            return -1;
        }
    }

    /* From here the connection is considered ready: with the new status,
     * poll() will use PQisBusy instead of PQconnectPoll.
     */
    self->status = CONN_STATUS_READY;

    if (conn_setup(self) == -1) {
        return -1;
    }

    return 0;
}

static int
_conn_async_connect(connectionObject *self)
{
    PGconn *pgconn;

    self->pgconn = pgconn = PQconnectStart(self->dsn);

    Dprintf("conn_connect: new postgresql connection at %p", pgconn);

    if (pgconn == NULL)
    {
        Dprintf("conn_connect: PQconnectStart(%s) FAILED", self->dsn);
        PyErr_SetString(OperationalError, "PQconnectStart() failed");
        return -1;
    }
    else if (PQstatus(pgconn) == CONNECTION_BAD)
    {
        Dprintf("conn_connect: PQconnectdb(%s) returned BAD", self->dsn);
        PyErr_SetString(OperationalError, PQerrorMessage(pgconn));
        return -1;
    }

    PQsetNoticeProcessor(pgconn, conn_notice_callback, (void*)self);

    /* Set the connection to nonblocking now. */
    if (pq_set_non_blocking(self, 1) != 0) {
        return -1;
    }

    /* The connection will be completed banging on poll():
     * First with _conn_poll_connecting() that will finish connection,
     * then with _conn_poll_setup_async() that will do the same job
     * of setup_async(). */

    return 0;
}

int
conn_connect(connectionObject *self, long int async)
{
    int rv;

    if (async == 1) {
      Dprintf("con_connect: connecting in ASYNC mode");
      rv = _conn_async_connect(self);
    }
    else {
      Dprintf("con_connect: connecting in SYNC mode");
      rv = _conn_sync_connect(self);
    }

    if (rv != 0) {
        /* connection failed, so let's close ourselves */
        self->closed = 2;
    }

    return rv;
}


/* poll during a connection attempt until the connection has established. */

static int
_conn_poll_connecting(connectionObject *self)
{
    int res = PSYCO_POLL_ERROR;
    const char *msg;

    Dprintf("conn_poll: poll connecting");
    switch (PQconnectPoll(self->pgconn)) {
    case PGRES_POLLING_OK:
        res = PSYCO_POLL_OK;
        break;
    case PGRES_POLLING_READING:
        res = PSYCO_POLL_READ;
        break;
    case PGRES_POLLING_WRITING:
        res = PSYCO_POLL_WRITE;
        break;
    case PGRES_POLLING_FAILED:
    case PGRES_POLLING_ACTIVE:
        msg = PQerrorMessage(self->pgconn);
        if (!(msg && *msg)) {
            msg = "asynchronous connection failed";
        }
        PyErr_SetString(OperationalError, msg);
        res = PSYCO_POLL_ERROR;
        break;
    }

    return res;
}


/* Advance to the next state after an attempt of flushing output */

static int
_conn_poll_advance_write(connectionObject *self)
{
    int res;
    int flush;

    Dprintf("conn_poll: poll writing");

    flush = PQflush(self->pgconn);
    Dprintf("conn_poll: PQflush() = %i", flush);

    switch (flush) {
    case  0:  /* success */
        /* we've finished pushing the query to the server. Let's start
          reading the results. */
        Dprintf("conn_poll: async_status -> ASYNC_READ");
        self->async_status = ASYNC_READ;
        res = PSYCO_POLL_READ;
        break;
    case  1:  /* would block */
        res = PSYCO_POLL_WRITE;
        break;
    case -1:  /* error */
        PyErr_SetString(OperationalError, PQerrorMessage(self->pgconn));
        res = PSYCO_POLL_ERROR;
        break;
    default:
        Dprintf("conn_poll: unexpected result from flush: %d", flush);
        res = PSYCO_POLL_ERROR;
        break;
    }
    return res;
}


/* Advance to the next state after reading results */

static int
_conn_poll_advance_read(connectionObject *self)
{
    int res;
    int busy;

    Dprintf("conn_poll: poll reading");

    busy = pq_get_result_async(self);

    switch (busy) {
    case 0: /* result is ready */
        Dprintf("conn_poll: async_status -> ASYNC_DONE");
        self->async_status = ASYNC_DONE;
        res = PSYCO_POLL_OK;
        break;
    case 1: /* result not ready: fd would block */
        res = PSYCO_POLL_READ;
        break;
    case -1: /* ouch, error */
        res = PSYCO_POLL_ERROR;
        break;
    default:
        Dprintf("conn_poll: unexpected result from pq_get_result_async: %d",
            busy);
        res = PSYCO_POLL_ERROR;
        break;
    }
    return res;
}


/* Poll the connection for the send query/retrieve result phase

  Advance the async_status (usually going WRITE -> READ -> DONE) but don't
  mess with the connection status. */

static int
_conn_poll_query(connectionObject *self)
{
    int res = PSYCO_POLL_ERROR;

    switch (self->async_status) {
    case ASYNC_WRITE:
        Dprintf("conn_poll: async_status = ASYNC_WRITE");
        res = _conn_poll_advance_write(self);
        break;

    case ASYNC_READ:
        Dprintf("conn_poll: async_status = ASYNC_READ");
        res = _conn_poll_advance_read(self);
        break;

    case ASYNC_DONE:
        Dprintf("conn_poll: async_status = ASYNC_DONE");
        /* We haven't asked anything: just check for notifications. */
        res = _conn_poll_advance_read(self);
        break;

    default:
        Dprintf("conn_poll: in unexpected async status: %d",
                self->async_status);
        res = PSYCO_POLL_ERROR;
        break;
    }

    return res;
}

/* Advance to the next state during an async connection setup
 *
 * If the connection is green, this is performed by the regular
 * sync code so the queries are sent by conn_setup() while in
 * CONN_STATUS_READY state.
 */
static int
_conn_poll_setup_async(connectionObject *self)
{
    int res = PSYCO_POLL_ERROR;

    switch (self->status) {
    case CONN_STATUS_CONNECTING:
        self->equote = conn_get_standard_conforming_strings(self->pgconn);
        self->protocol = conn_get_protocol_version(self->pgconn);
        self->server_version = conn_get_server_version(self->pgconn);
        if (3 != self->protocol) {
            PyErr_SetString(InterfaceError, "only protocol 3 supported");
            break;
        }
        if (0 > conn_read_encoding(self, self->pgconn)) {
            break;
        }
        if (0 > conn_setup_cancel(self, self->pgconn)) {
            return -1;
        }

        /* asynchronous connections always use isolation level 0, the user is
         * expected to manage the transactions himself, by sending
         * (asynchronously) BEGIN and COMMIT statements.
         */
        self->autocommit = 1;

        /* If the datestyle is ISO or anything else good,
         * we can skip the CONN_STATUS_DATESTYLE step.
         * Note that we cannot change the datestyle on a replication
         * connection.
         */
        if (!dsn_has_replication(self->dsn) && !conn_is_datestyle_ok(self->pgconn)) {
            Dprintf("conn_poll: status -> CONN_STATUS_DATESTYLE");
            self->status = CONN_STATUS_DATESTYLE;
            if (0 == pq_send_query(self, psyco_datestyle)) {
                PyErr_SetString(OperationalError, PQerrorMessage(self->pgconn));
                break;
            }
            Dprintf("conn_poll: async_status -> ASYNC_WRITE");
            self->async_status = ASYNC_WRITE;
            res = PSYCO_POLL_WRITE;
        }
        else {
            Dprintf("conn_poll: status -> CONN_STATUS_READY");
            self->status = CONN_STATUS_READY;
            res = PSYCO_POLL_OK;
        }
        break;

    case CONN_STATUS_DATESTYLE:
        res = _conn_poll_query(self);
        if (res == PSYCO_POLL_OK) {
            res = PSYCO_POLL_ERROR;
            if (self->pgres == NULL
                    || PQresultStatus(self->pgres) != PGRES_COMMAND_OK ) {
                PyErr_SetString(OperationalError, "can't set datestyle to ISO");
                break;
            }
            CLEARPGRES(self->pgres);

            Dprintf("conn_poll: status -> CONN_STATUS_READY");
            self->status = CONN_STATUS_READY;
            res = PSYCO_POLL_OK;
        }
        break;
    }
    return res;
}


static cursorObject *
_conn_get_async_cursor(connectionObject *self) {
    PyObject *py_curs;

    if (!(py_curs = PyWeakref_GetObject(self->async_cursor))) {
        PyErr_SetString(PyExc_SystemError,
            "got null dereferencing cursor weakref");
        goto error;
    }
    if (Py_None == py_curs) {
        PyErr_SetString(InterfaceError,
            "the asynchronous cursor has disappeared");
        goto error;
    }

    Py_INCREF(py_curs);
    return (cursorObject *)py_curs;

error:
    pq_clear_async(self);
    return NULL;
}

/* conn_poll - Main polling switch
 *
 * The function is called in all the states and connection types and invokes
 * the right "next step".
 */

int
conn_poll(connectionObject *self)
{
    int res = PSYCO_POLL_ERROR;
    Dprintf("conn_poll: status = %d", self->status);

    switch (self->status) {
    case CONN_STATUS_SETUP:
        Dprintf("conn_poll: status -> CONN_STATUS_SETUP");
        self->status = CONN_STATUS_CONNECTING;
        res = PSYCO_POLL_WRITE;
        break;

    case CONN_STATUS_CONNECTING:
        Dprintf("conn_poll: status -> CONN_STATUS_CONNECTING");
        res = _conn_poll_connecting(self);
        if (res == PSYCO_POLL_OK && self->async) {
            res = _conn_poll_setup_async(self);
        }
        break;

    case CONN_STATUS_DATESTYLE:
        Dprintf("conn_poll: status -> CONN_STATUS_DATESTYLE");
        res = _conn_poll_setup_async(self);
        break;

    case CONN_STATUS_READY:
    case CONN_STATUS_BEGIN:
    case CONN_STATUS_PREPARED:
        Dprintf("conn_poll: status -> CONN_STATUS_*");
        res = _conn_poll_query(self);

        if (res == PSYCO_POLL_OK && self->async && self->async_cursor) {
            cursorObject *curs;

            /* An async query has just finished: parse the tuple in the
             * target cursor. */
            if (!(curs = _conn_get_async_cursor(self))) {
                res = PSYCO_POLL_ERROR;
                break;
            }

            curs_set_result(curs, self->pgres);
            self->pgres = NULL;

            /* fetch the tuples (if there are any) and build the result. We
             * don't care if pq_fetch return 0 or 1, but if there was an error,
             * we want to signal it to the caller. */
            if (pq_fetch(curs, 0) == -1) {
               res = PSYCO_POLL_ERROR;
            }

            /* We have finished with our async_cursor */
            Py_DECREF(curs);
            Py_CLEAR(self->async_cursor);
        }
        break;

    default:
        Dprintf("conn_poll: in unexpected state");
        res = PSYCO_POLL_ERROR;
    }

    Dprintf("conn_poll: returning %d", res);
    return res;
}

/* conn_close - do anything needed to shut down the connection */

void
conn_close(connectionObject *self)
{
    /* a connection with closed == 2 still requires cleanup */
    if (self->closed == 1) {
        return;
    }

    /* sets this connection as closed even for other threads; */
    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);

    conn_close_locked(self);

    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;
}

/* conn_close_locked - shut down the connection with the lock already taken */

void conn_close_locked(connectionObject *self)
{
    if (self->closed == 1) {
        return;
    }

    /* We used to call pq_abort_locked here, but the idea of issuing a
     * rollback on close/GC has been considered inappropriate.
     *
     * Dropping the connection on the server has the same effect as the
     * transaction is automatically rolled back. Some middleware, such as
     * PgBouncer, have problem with connections closed in the middle of the
     * transaction though: to avoid these problems the transaction should be
     * closed only in status CONN_STATUS_READY.
     */
    self->closed = 1;

    /* we need to check the value of pgconn, because we get called even when
     * the connection fails! */
    if (self->pgconn) {
        PQfinish(self->pgconn);
        self->pgconn = NULL;
        Dprintf("conn_close: PQfinish called");
    }
}

/* conn_commit - commit on a connection */

RAISES_NEG int
conn_commit(connectionObject *self)
{
    int res;

    res = pq_commit(self);
    return res;
}

/* conn_rollback - rollback a connection */

RAISES_NEG int
conn_rollback(connectionObject *self)
{
    int res;

    res = pq_abort(self);
    return res;
}


/* Change the state of the session */
RAISES_NEG int
conn_set_session(connectionObject *self, int autocommit,
        int isolevel, int readonly, int deferrable)
{
    int rv = -1;
    int want_autocommit = autocommit == SRV_STATE_UNCHANGED ?
        self->autocommit : autocommit;

    if (deferrable != SRV_STATE_UNCHANGED && self->server_version < 90100) {
        PyErr_SetString(ProgrammingError,
            "the 'deferrable' setting is only available"
            " from PostgreSQL 9.1");
        goto exit;
    }

    /* Promote an isolation level to one of the levels supported by the server */
    if (self->server_version < 80000) {
        if (isolevel == ISOLATION_LEVEL_READ_UNCOMMITTED) {
            isolevel = ISOLATION_LEVEL_READ_COMMITTED;
        }
        else if (isolevel == ISOLATION_LEVEL_REPEATABLE_READ) {
            isolevel = ISOLATION_LEVEL_SERIALIZABLE;
        }
    }

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);

    if (want_autocommit) {
        /* we are or are going in autocommit state, so no BEGIN will be issued:
         * configure the session with the characteristics requested */
        if (isolevel != SRV_STATE_UNCHANGED) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_isolation", srv_isolevels[isolevel],
                    &_save)) {
                goto endlock;
            }
        }
        if (readonly != SRV_STATE_UNCHANGED) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_read_only", srv_state_guc[readonly],
                    &_save)) {
                goto endlock;
            }
        }
        if (deferrable != SRV_STATE_UNCHANGED) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_deferrable", srv_state_guc[deferrable],
                    &_save)) {
                goto endlock;
            }
        }
    }
    else if (self->autocommit) {
        /* we are moving from autocommit to not autocommit, so revert the
         * characteristics to defaults to let BEGIN do its work */
        if (self->isolevel != ISOLATION_LEVEL_DEFAULT) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_isolation", "default",
                    &_save)) {
                goto endlock;
            }
        }
        if (self->readonly != STATE_DEFAULT) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_read_only", "default",
                    &_save)) {
                goto endlock;
            }
        }
        if (self->server_version >= 90100 && self->deferrable != STATE_DEFAULT) {
            if (0 > pq_set_guc_locked(self,
                    "default_transaction_deferrable", "default",
                    &_save)) {
                goto endlock;
            }
        }
    }

    if (autocommit != SRV_STATE_UNCHANGED) {
        self->autocommit = autocommit;
    }
    if (isolevel != SRV_STATE_UNCHANGED) {
        self->isolevel = isolevel;
    }
    if (readonly != SRV_STATE_UNCHANGED) {
        self->readonly = readonly;
    }
    if (deferrable != SRV_STATE_UNCHANGED) {
        self->deferrable = deferrable;
    }
    rv = 0;

endlock:
    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;

    if (rv < 0) {
        pq_complete_error(self);
        goto exit;
    }

    Dprintf(
        "conn_set_session: autocommit %d, isolevel %d, readonly %d, deferrable %d",
        autocommit, isolevel, readonly, deferrable);


exit:
    return rv;
}


/* conn_set_client_encoding - switch client encoding on connection */

RAISES_NEG int
conn_set_client_encoding(connectionObject *self, const char *pgenc)
{
    int res = -1;
    char *clean_enc = NULL;

    /* We must know what python encoding this encoding is. */
    if (0 > clear_encoding_name(pgenc, &clean_enc)) { goto exit; }

    /* If the current encoding is equal to the requested one we don't
       issue any query to the backend */
    if (strcmp(self->encoding, clean_enc) == 0) return 0;

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);

    /* abort the current transaction, to set the encoding ouside of
       transactions */
    if ((res = pq_abort_locked(self, &_save))) {
        goto endlock;
    }

    if ((res = pq_set_guc_locked(self, "client_encoding", clean_enc, &_save))) {
        goto endlock;
    }

endlock:
    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;

    if (res < 0) {
        pq_complete_error(self);
        goto exit;
    }

    res = conn_store_encoding(self, pgenc);

    Dprintf("conn_set_client_encoding: encoding set to %s", self->encoding);

exit:
    PyMem_Free(clean_enc);

    return res;
}


/* conn_tpc_begin -- begin a two-phase commit.
 *
 * The state of a connection in the middle of a TPC is exactly the same
 * of a normal transaction, in CONN_STATUS_BEGIN, but with the tpc_xid
 * member set to the xid used. This allows to reuse all the code paths used
 * in regular transactions, as PostgreSQL won't even know we are in a TPC
 * until PREPARE. */

RAISES_NEG int
conn_tpc_begin(connectionObject *self, xidObject *xid)
{
    Dprintf("conn_tpc_begin: starting transaction");

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);

    if (pq_begin_locked(self, &_save) < 0) {
        pthread_mutex_unlock(&(self->lock));
        Py_BLOCK_THREADS;
        pq_complete_error(self);
        return -1;
    }

    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;

    /* The transaction started ok, let's store this xid. */
    Py_INCREF(xid);
    self->tpc_xid = xid;

    return 0;
}


/* conn_tpc_command -- run one of the TPC-related PostgreSQL commands.
 *
 * The function doesn't change the connection state as it can be used
 * for many commands and for recovered transactions. */

RAISES_NEG int
conn_tpc_command(connectionObject *self, const char *cmd, xidObject *xid)
{
    PyObject *tid = NULL;
    const char *ctid;
    int rv = -1;

    Dprintf("conn_tpc_command: %s", cmd);

    /* convert the xid into PostgreSQL transaction id while keeping the GIL */
    if (!(tid = psyco_ensure_bytes(xid_get_tid(xid)))) { goto exit; }
    if (!(ctid = Bytes_AsString(tid))) { goto exit; }

    Py_BEGIN_ALLOW_THREADS;
    pthread_mutex_lock(&self->lock);

    if (0 > (rv = pq_tpc_command_locked(self, cmd, ctid, &_save))) {
        pthread_mutex_unlock(&self->lock);
        Py_BLOCK_THREADS;
        pq_complete_error(self);
        goto exit;
    }

    pthread_mutex_unlock(&self->lock);
    Py_END_ALLOW_THREADS;

exit:
    Py_XDECREF(tid);
    return rv;
}

/* conn_tpc_recover -- return a list of pending TPC Xid */

PyObject *
conn_tpc_recover(connectionObject *self)
{
    int status;
    PyObject *xids = NULL;
    PyObject *rv = NULL;
    PyObject *tmp;

    /* store the status to restore it. */
    status = self->status;

    if (!(xids = xid_recover((PyObject *)self))) { goto exit; }

    if (status == CONN_STATUS_READY && self->status == CONN_STATUS_BEGIN) {
        /* recover began a transaction: let's abort it. */
        if (!(tmp = PyObject_CallMethod((PyObject *)self, "rollback", NULL))) {
            goto exit;
        }
        Py_DECREF(tmp);
    }

    /* all fine */
    rv = xids;
    xids = NULL;

exit:
    Py_XDECREF(xids);

    return rv;

}


void
conn_set_result(connectionObject *self, PGresult *pgres)
{
    PQclear(self->pgres);
    self->pgres = pgres;
}


void
conn_set_error(connectionObject *self, const char *msg)
{
    if (self->error) {
        free(self->error);
        self->error = NULL;
    }
    if (msg && *msg) {
        self->error = strdup(msg);
    }
}