gistfile1.c

static VALUE
pgconn_async_exec(obj, str)
    VALUE obj, str;
{
    PGconn *conn = get_pgconn(obj);
    PGresult *result;
    char *msg;

    int cs;
    int ret;
    fd_set rset;

    Check_Type(str, T_STRING);
        
    while ((result = PQgetResult(conn)) != NULL) {
        PQclear(result);
    }

    if (!PQsendQuery(conn, RSTRING_PTR(str))) {
        rb_raise(rb_ePGError, PQerrorMessage(conn));
    }

    cs = PQsocket(conn);
    for(;;) {
        FD_ZERO(&rset);
        FD_SET(cs, &rset);
        ret = rb_thread_select(cs + 1, &rset, NULL, NULL, NULL);
        if (ret < 0) {
            rb_sys_fail(0);
        }
                
        if (ret == 0) {
            continue;
        }

        if (PQconsumeInput(conn) == 0) {
            rb_raise(rb_ePGError, PQerrorMessage(conn));
        }

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

    result = PQgetResult(conn);

    if (!result) {
        rb_raise(rb_ePGError, PQerrorMessage(conn));
    }

    switch (PQresultStatus(result)) {
    case PGRES_TUPLES_OK:
    case PGRES_COPY_OUT:
    case PGRES_COPY_IN:
    case PGRES_EMPTY_QUERY:
    case PGRES_COMMAND_OK:      /* no data will be received */
      return pgresult_new(result);

    case PGRES_BAD_RESPONSE:
    case PGRES_FATAL_ERROR:
    case PGRES_NONFATAL_ERROR:
      msg = RSTRING_PTR(rb_str_new2(PQresultErrorMessage(result)));
      break;
    default:
      msg = "internal error : unknown result status.";
      break;
    }
    PQclear(result);
    rb_raise(rb_ePGError, msg);
}