SWITCH_DECLARE(switch_pgsql_status_t) switch_pgsql_SQLEndTran(switch_pgsql_handle_t *handle, switch_bool_t commit)
{
#ifdef SWITCH_HAVE_PGSQL
    char * err_str = NULL;
    if (commit) {
        if(!PQsendQuery(handle->con, "COMMIT")) {
            err_str = switch_pgsql_handle_get_error(handle);
            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Could not commit transaction: %s\n", err_str);
            switch_safe_free(err_str);
            return SWITCH_PGSQL_FAIL;
        }
    } else {
        if(!PQsendQuery(handle->con, "ROLLBACK")) {
            err_str = switch_pgsql_handle_get_error(handle);
            switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_CRIT, "Could not rollback transaction: %s\n", err_str);
            switch_safe_free(err_str);
            return SWITCH_PGSQL_FAIL;
        }
    }
    handle->in_txn = SWITCH_FALSE;
    return SWITCH_PGSQL_SUCCESS;
#else
    return (switch_pgsql_status_t) SWITCH_FALSE;
#endif
}

PGresult * do_postgres_cCommand_execute_async(VALUE self, VALUE connection, PGconn *db, VALUE query) {
  PGresult *response;
  char* str = StringValuePtr(query);

  while ((response = PQgetResult(db))) {
    PQclear(response);
  }

  struct timeval start;
  int retval;

  gettimeofday(&start, NULL);
  retval = PQsendQuery(db, str);

  if (!retval) {
    if (PQstatus(db) != CONNECTION_OK) {
      PQreset(db);

      if (PQstatus(db) == CONNECTION_OK) {
        retval = PQsendQuery(db, str);
      }
      else {
        do_postgres_full_connect(connection, db);
        retval = PQsendQuery(db, str);
      }
    }

    if (!retval) {
      rb_raise(eDO_ConnectionError, "%s", PQerrorMessage(db));
    }
  }

  int socket_fd = PQsocket(db);
  fd_set rset;

  while (1) {
    FD_ZERO(&rset);
    FD_SET(socket_fd, &rset);
    retval = rb_thread_select(socket_fd + 1, &rset, NULL, NULL, NULL);

    if (retval < 0) {
      rb_sys_fail(0);
    }

    if (retval == 0) {
      continue;
    }

    if (PQconsumeInput(db) == 0) {
      rb_raise(eDO_ConnectionError, "%s", PQerrorMessage(db));
    }

    if (PQisBusy(db) == 0) {
      break;
    }
  }

  data_objects_debug(connection, query, &start);
  return PQgetResult(db);
}

 bool executeAsyncQuery(std::string const& sql, int type = 0)
 {
     int result = 0;
     if (type == 1)
     {
         result = PQsendQueryParams(conn_,sql.c_str(), 0, 0, 0, 0, 0, 1);
     }
     else
     {
         result = PQsendQuery(conn_, sql.c_str());
     }
     if (result != 1)
     {
         std::string err_msg = "Postgis Plugin: ";
         err_msg += status();
         err_msg += "in executeAsyncQuery Full sql was: '";
         err_msg += sql;
         err_msg += "'\n";
         clearAsyncResult(PQgetResult(conn_));
         close();
         throw mapnik::datasource_exception(err_msg);
     }
     pending_ = true;
     return result;
 }

static awk_value_t *
do_pg_sendquery(int nargs, awk_value_t *result)
{
  PGconn *conn;
  awk_value_t command;
  int res;

  if (do_lint && (nargs > 2))
    lintwarn(ext_id, _("pg_sendquery: called with too many arguments"));

  if (!(conn = find_handle(conns, 0))) {
    set_ERRNO(_("pg_sendquery called with unknown connection handle"));
    RET_NUM(0);
  }

  if (!get_argument(1, AWK_STRING, &command)) {
    set_ERRNO(_("pg_sendquery 2nd argument should be a string"));
    RET_NUM(0);
  }

  res = PQsendQuery(conn, command.str_value.str);
  if (!res)
    /* connection is probably bad */
    set_ERRNO(PQerrorMessage(conn));
  RET_NUM(res);
}

static void pq_event(evutil_socket_t fd, short event, void *arg) {
  struct connection_struct* database = (struct connection_struct*) arg;
  if (database->queries) {
    if (database->queries->sent == 0) {
      PQsendQuery(database->conn, database->queries->query);
      database->queries->sent = 1;
    }
    if (PQconsumeInput(database->conn) && !PQisBusy(database->conn)) {
      PGresult* res = PQgetResult(database->conn);
      while (res) {
        if (database->queries->callback)
          database->queries->callback(res, database->queries->context, database->queries->query);
        if (database->report_errors && PQresultStatus(res) != PGRES_COMMAND_OK)
          fprintf(stderr, "Query: '%s' returned error\n\t%s\n", database->queries->query, PQresultErrorMessage(res));
        PQclear(res);
        res = PQgetResult(database->conn);
      }
      database->query_count--;
      struct query_struct* old = database->queries;
      database->queries = database->queries->next;
      free(old->query);
      free(old);
      pq_event(fd, event, arg);
    }
  }
}

/* load rows directly from network buffer */
static void exec_query_zero_copy(struct Context *ctx, const char *q)
{
	PGconn *db = ctx->db;
	PGresult *r;
	ExecStatusType s;
	PGdataValue *cols;

	ctx->count = 0;

	if (!PQsendQuery(db, q))
		die(db, "PQsendQuery");

	if (!PQsetSingleRowMode(db))
		die(NULL, "PQsetSingleRowMode");

	/* loop until all resultset is done */
	while (PQgetRowData(db, &r, &cols)) {
		proc_row_zcopy(ctx, r, cols);
	}

	/* get final result */
	r = PQgetResult(db);
	s = PQresultStatus(r);
	switch (s) {
	case PGRES_TUPLES_OK:
		//printf("query successful, got %d rows\n", ctx->count);
		ctx->count = 0;
		break;
	default:
		printf("result: %s\n", PQresStatus(s));
		break;
	}
	PQclear(r);
}

/*
 * Send a query and wait for the results by using the asynchronous libpq
 * functions and the backend version of select().
 *
 * We must not use the regular blocking libpq functions like PQexec()
 * since they are uninterruptible by signals on some platforms, such as
 * Windows.
 *
 * We must also not use vanilla select() here since it cannot handle the
 * signal emulation layer on Windows.
 *
 * The function is modeled on PQexec() in libpq, but only implements
 * those parts that are in use in the walreceiver.
 *
 * Queries are always executed on the connection in streamConn.
 */
static PGresult *
libpqrcv_PQexec(const char *query)
{
	PGresult   *result = NULL;
	PGresult   *lastResult = NULL;

	/*
	 * PQexec() silently discards any prior query results on the connection.
	 * This is not required for walreceiver since it's expected that walsender
	 * won't generate any such junk results.
	 */

	/*
	 * Submit a query. Since we don't use non-blocking mode, this also can
	 * block. But its risk is relatively small, so we ignore that for now.
	 */
	if (!PQsendQuery(streamConn, query))
		return NULL;

	for (;;)
	{
		/*
		 * Receive data until PQgetResult is ready to get the result without
		 * blocking.
		 */
		while (PQisBusy(streamConn))
		{
			/*
			 * We don't need to break down the sleep into smaller increments,
			 * and check for interrupts after each nap, since we can just
			 * elog(FATAL) within SIGTERM signal handler if the signal arrives
			 * in the middle of establishment of replication connection.
			 */
			if (!libpq_select(-1))
				continue;		/* interrupted */
			if (PQconsumeInput(streamConn) == 0)
				return NULL;	/* trouble */
		}

		/*
		 * Emulate the PQexec()'s behavior of returning the last result when
		 * there are many. Since walsender will never generate multiple
		 * results, we skip the concatenation of error messages.
		 */
		result = PQgetResult(streamConn);
		if (result == NULL)
			break;				/* query is complete */

		PQclear(lastResult);
		lastResult = result;

		if (PQresultStatus(lastResult) == PGRES_COPY_IN ||
			PQresultStatus(lastResult) == PGRES_COPY_OUT ||
			PQresultStatus(lastResult) == PGRES_COPY_BOTH ||
			PQstatus(streamConn) == CONNECTION_BAD)
			break;
	}

	return lastResult;
}

bool
pgut_send(PGconn* conn, const char *query, int nParams, const char **params)
{
	int			res;

	CHECK_FOR_INTERRUPTS();

	/* write query to elog if debug */
	if (pgut_echo)
		echo_query(query, nParams, params);

	if (conn == NULL)
	{
		ereport(ERROR,
			(errcode(E_PG_COMMAND),
			 errmsg("not connected")));
		return false;
	}

	if (nParams == 0)
		res = PQsendQuery(conn, query);
	else
		res = PQsendQueryParams(conn, query, nParams, NULL, params, NULL, NULL, 0);

	if (res != 1)
	{
		ereport(ERROR,
			(errcode(E_PG_COMMAND),
			 errmsg("query failed: %s", PQerrorMessage(conn)),
			 errdetail("query was: %s", query)));
		return false;
	}

	return true;
}

/*
 * Execute a vacuum/analyze command to the server.
 *
 * Result status is checked only if 'async' is false.
 */
static void
run_vacuum_command(PGconn *conn, const char *sql, bool echo,
				   const char *dbname, const char *table,
				   const char *progname, bool async)
{
	if (async)
	{
		if (echo)
			printf("%s\n", sql);

		PQsendQuery(conn, sql);
	}
	else if (!executeMaintenanceCommand(conn, sql, echo))
	{
		if (table)
			fprintf(stderr,
			_("%s: vacuuming of table \"%s\" in database \"%s\" failed: %s"),
					progname, table, dbname, PQerrorMessage(conn));
		else
			fprintf(stderr, _("%s: vacuuming of database \"%s\" failed: %s"),
					progname, dbname, PQerrorMessage(conn));
		PQfinish(conn);
		exit(1);
	}
}

		sentquery(const authpgsql_connection &conn,
			  const std::string &query)
			: status(PQsendQuery(conn.pgconn, query.c_str()))
		{
			if (status == 0)
				DPRINTF("PQsendQuery failed: %s",
					PQerrorMessage(conn.pgconn));
		}

/* check the PQStatus and try to 'select 1' to confirm good connection */
bool
is_pgup(PGconn *conn, int timeout)
{
	char		sqlquery[QUERY_STR_LEN];

	/* Check the connection status twice in case it changes after reset */
	bool		twice = false;

	/* Check the connection status twice in case it changes after reset */
	for (;;)
	{
		if (PQstatus(conn) != CONNECTION_OK)
		{
			if (twice)
				return false;
			PQreset(conn);		/* reconnect */
			twice = true;
		}
		else
		{
			/*
			 * Send a SELECT 1 just to check if the connection is OK
			 */
			if (!cancel_query(conn, timeout))
				goto failed;
			if (wait_connection_availability(conn, timeout) != 1)
				goto failed;

			sqlquery_snprintf(sqlquery, "SELECT 1");
			if (PQsendQuery(conn, sqlquery) == 0)
			{
				log_warning(_("PQsendQuery: Query could not be sent to primary. %s\n"),
							PQerrorMessage(conn));
				goto failed;
			}
			if (wait_connection_availability(conn, timeout) != 1)
				goto failed;

			break;

	failed:

			/*
			 * we need to retry, because we might just have lost the
			 * connection once
			 */
			if (twice)
				return false;
			PQreset(conn);		/* reconnect */
			twice = true;
		}
	}
	return true;
}

/*
 * Submit a query and wait for the result.
 *
 * This function is interruptible by signals.
 *
 * Caller is responsible for the error handling on the result.
 */
PGresult *
pgfdw_exec_query(PGconn *conn, const char *query)
{
	/*
	 * Submit a query.  Since we don't use non-blocking mode, this also can
	 * block.  But its risk is relatively small, so we ignore that for now.
	 */
	if (!PQsendQuery(conn, query))
		pgfdw_report_error(ERROR, NULL, conn, false, query);

	/* Wait for the result. */
	return pgfdw_get_result(conn, query);
}

ngx_int_t
ngx_postgres_upstream_send_query(ngx_http_request_t *r, ngx_connection_t *pgxc,
    ngx_postgres_upstream_peer_data_t *pgdt)
{
    ngx_postgres_loc_conf_t  *pglcf;
    ngx_int_t                 pgrc;
    u_char                   *query;

    dd("entering");

    pglcf = ngx_http_get_module_loc_conf(r, ngx_postgres_module);

    query = ngx_pnalloc(r->pool, pgdt->query.len + 1);
    if (query == NULL) {
        dd("returning NGX_ERROR");
        return NGX_ERROR;
    }

    (void) ngx_cpystrn(query, pgdt->query.data, pgdt->query.len + 1);

    dd("sending query: %s", query);

    if (pglcf->output_binary) {
        pgrc = PQsendQueryParams(pgdt->pgconn, (const char *) query,
                                 0, NULL, NULL, NULL, NULL, /* binary */ 1);
    } else {
        pgrc = PQsendQuery(pgdt->pgconn, (const char *) query);
    }

    if (pgrc == 0) {
        dd("sending query failed");
        ngx_log_error(NGX_LOG_ERR, pgxc->log, 0,
                      "postgres: sending query failed: %s",
                      PQerrorMessage(pgdt->pgconn));

        dd("returning NGX_ERROR");
        return NGX_ERROR;
    }

    /* set result timeout */
    ngx_add_timer(pgxc->read, r->upstream->conf->read_timeout);

    dd("query sent successfully");

    pgxc->log->action = "waiting for result from PostgreSQL database";
    pgdt->state = state_db_get_result;

    dd("returning NGX_DONE");
    return NGX_DONE;
}

bool PostgreSQLConnection::_Query(const char* sql)
{
	if (!_pgConn)
		return false;

	int returnVal = PQsendQuery(_pgConn,sql);
	if (!returnVal)
	{
		bool connLost = _ConnectionLost();
		throw SqlException(returnVal,lastErrorDescr(),"Query",connLost,connLost,sql);
	}

	return true;
}

CatchChallenger::DatabaseBase::CallBack * EpollPostgresql::asyncRead(const std::string &query,void * returnObject,CallBackDatabase method)
{
    if(conn==NULL)
    {
        std::cerr << "pg not connected" << std::endl;
        return NULL;
    }
    tempCallback.object=returnObject;
    tempCallback.method=method;
    PreparedStatement tempQuery;
    #if defined(CATCHCHALLENGER_DB_PREPAREDSTATEMENT)
    tempQuery.id=NULL;
    tempQuery.paramValues=NULL;
    tempQuery.paramValuesBuffer=NULL;
    tempQuery.paramValuesCount=0;
    #endif
    tempQuery.query=query;
    if(queue.size()>0 || result!=NULL)
    {
        if(queue.size()>=maxDbQueries)
        {
            std::cerr << "pg queue full" << std::endl;
            return NULL;
        }
        queue.push_back(tempCallback);
        queriesList.push_back(tempQuery);
        return &queue.back();
    }
    start = std::chrono::high_resolution_clock::now();
    queriesList.push_back(tempQuery);
    #ifdef CATCHCHALLENGER_EXTRA_CHECK
    const int &stringlen=static_cast<int>(query.size());
    if(stringlen==0)
    {
        std::cerr << "query " << query << ", stringlen==0" << std::endl;
        abort();
    }
    #endif
    #ifdef DEBUG_MESSAGE_CLIENT_SQL
    std::cout << simplifiedstrCoPG << ", query " << query << " at " << std::string(__FILE__) << ":" << std::to_string(__LINE__) << std::endl;
    #endif
    const int &query_id=PQsendQuery(conn,query.c_str());
    if(query_id==0)
    {
        std::cerr << "query send failed: " << errorMessage() << std::endl;
        return NULL;
    }
    queue.push_back(tempCallback);
    return &queue.back();
}

/** Returns number of records returned by given command. */
int xdb_row_count(char *command)
{
    int errno = 0, tuples = 0;
    PGconn *conn = xdb_connect();
    PQsendQuery(conn, command);
    PGresult *res = PQgetResult(conn);
    errno = PQresultStatus(res); 
    if (errno > PGRES_TUPLES_OK) { 
	fprintf(stderr, "error: DATABASE command failed: %i\n", errno);
	return 0;
    }
    tuples = PQntuples(res);
    PQfinish(conn);
    return tuples;
}

static void 
evpg_make_evquery(int sock, short which, void **data)
{
    struct evpg_db_node *dbnode;
    struct evpg_cfg *config;
    const char *querystr;
    void (*cb)(PGresult *, void *);
    void *usrdata;
    struct event *event;

    config   = data[0];
    querystr = data[1];
    cb       = data[2];
    usrdata  = data[3];
    dbnode   = data[4];
    event    = data[5];

    if (!dbnode)
    {
       	if (!(dbnode = evpg_snatch_connection(config)))
	{
	    event_set(event, 0, EV_WRITE, (void *)evpg_make_evquery, data);
	    event_add(event, 0);
	    return;
	}
    }

    PQsendQuery(dbnode->dbconn, querystr);

    if (PQstatus(dbnode->dbconn) != CONNECTION_OK)
    {
	cb(NULL, usrdata);
	evpg_set_ready(config, dbnode);
	free(event);
	free(data);
    }

    data[4] = dbnode;

    evpg_set_active(config, dbnode);

    event_set(&dbnode->event, PQsocket(dbnode->dbconn), EV_WRITE, 
	    (void *)evpg_query_finished, data);
    event_add(&dbnode->event, 0);
}

CAMLprim value PQsendQueryParams_stub(
  value v_conn, value v_query, value v_params, value v_binary_params)
{
  PGconn *conn = get_conn(v_conn);
  const char *query = String_val(v_query);
  size_t nparams = Wosize_val(v_params);
  const char * const *params = copy_params_shallow(v_params, nparams);
  int *lengths, *formats, res;
  copy_binary_params(v_params, v_binary_params, nparams, &formats, &lengths);
  res =
    (nparams == 0)
      ? PQsendQuery(conn, query)
      : PQsendQueryParams(
          conn, query, nparams, NULL, params, lengths, formats, 0);
  free_binary_params(formats, lengths);
  free_params_shallow(params, nparams);
  return Val_int(res);
}

    bool Connection::ping()
    {
      log_debug("ping()");

      if (PQsendQuery(conn, "select 1") == 0)
      {
        log_debug("failed to send statement \"select 1\" to database in Connection::ping()");
        return false;
      }

      while (true)
      {
        struct pollfd fd;
        fd.fd = PQsocket(conn);
        fd.events = POLLIN;
        log_debug("wait for input on fd " << fd.fd);
        if (::poll(&fd, 1, 10000) != 1)
        {
          log_debug("no data received in Connection::ping()");
          return false;
        }

        log_debug("consumeInput");
        if (PQconsumeInput(conn) == 0)
        {
          log_debug("PQconsumeInput failed in Connection::ping()");
          return false;
        }

        log_debug("check PQisBusy");
        while (PQisBusy(conn) == 0)
        {
          log_debug("PQgetResult");
          PGresult* result = PQgetResult(conn);

          log_debug("PQgetResult => " << static_cast<void*>(result));
          if (result == 0)
            return true;

          log_debug("PQfree");
          PQclear(result);
        }
      }
    }

bool
pgut_send(PGconn* conn, const char *query, int nParams, const char **params)
{
	int			res;
	int			i;

	if (interrupted && !in_cleanup)
		ereport(FATAL,
			(errcode(ERROR_INTERRUPTED),
			 errmsg("interrupted")));

	/* write query to elog with DEBUG level */
	if (strchr(query, '\n'))
		elog(DEBUG, "(query)\n%s", query);
	else
		elog(DEBUG, "(query) %s", query);
	for (i = 0; i < nParams; i++)
		elog(DEBUG, "\t(param:%d) = %s", i, params[i] ? params[i] : "(null)");

	if (conn == NULL)
	{
		ereport(ERROR,
			(errcode(ERROR_PG_CONNECT),
			 errmsg("not connected")));
		return false;
	}

	if (nParams == 0)
		res = PQsendQuery(conn, query);
	else
		res = PQsendQueryParams(conn, query, nParams, NULL, params, NULL, NULL, 0);

	if (res != 1)
	{
		ereport(ERROR,
			(errcode(ERROR_PG_COMMAND),
			 errmsg("query failed: %squery was: %s",
			PQerrorMessage(conn), query)));
		return false;
	}

	return true;
}