*** a/src/interfaces/libpq/exports.txt --- b/src/interfaces/libpq/exports.txt *************** *** 160,162 **** PQconnectStartParams 157 --- 160,165 ---- PQping 158 PQpingParams 159 PQlibVersion 160 + PQsetRowProcessor 161 + PQgetRowProcessor 162 + PQskipResult 163 *** a/src/interfaces/libpq/fe-connect.c --- b/src/interfaces/libpq/fe-connect.c *************** *** 2693,2698 **** makeEmptyPGconn(void) --- 2693,2701 ---- conn->wait_ssl_try = false; #endif + /* set default row processor */ + PQsetRowProcessor(conn, NULL, NULL); + /* * We try to send at least 8K at a time, which is the usual size of pipe * buffers on Unix systems. That way, when we are sending a large amount *************** *** 2711,2718 **** makeEmptyPGconn(void) --- 2714,2726 ---- initPQExpBuffer(&conn->errorMessage); initPQExpBuffer(&conn->workBuffer); + /* set up initial row buffer */ + conn->rowBufLen = 32; + conn->rowBuf = (PGrowValue *)malloc(conn->rowBufLen * sizeof(PGrowValue)); + if (conn->inBuffer == NULL || conn->outBuffer == NULL || + conn->rowBuf == NULL || PQExpBufferBroken(&conn->errorMessage) || PQExpBufferBroken(&conn->workBuffer)) { *************** *** 2814,2819 **** freePGconn(PGconn *conn) --- 2822,2829 ---- free(conn->inBuffer); if (conn->outBuffer) free(conn->outBuffer); + if (conn->rowBuf) + free(conn->rowBuf); termPQExpBuffer(&conn->errorMessage); termPQExpBuffer(&conn->workBuffer); *************** *** 5078,5080 **** PQregisterThreadLock(pgthreadlock_t newhandler) --- 5088,5091 ---- return prev; } + *** a/src/interfaces/libpq/fe-exec.c --- b/src/interfaces/libpq/fe-exec.c *************** *** 66,71 **** static PGresult *PQexecFinish(PGconn *conn); --- 66,72 ---- static int PQsendDescribe(PGconn *conn, char desc_type, const char *desc_target); static int check_field_number(const PGresult *res, int field_num); + static int pqAddRow(PGresult *res, PGrowValue *columns, void *param); /* ---------------- *************** *** 701,707 **** pqClearAsyncResult(PGconn *conn) if (conn->result) PQclear(conn->result); conn->result = NULL; - conn->curTuple = NULL; } /* --- 702,707 ---- *************** *** 756,762 **** pqPrepareAsyncResult(PGconn *conn) */ res = conn->result; conn->result = NULL; /* handing over ownership to caller */ - conn->curTuple = NULL; /* just in case */ if (!res) res = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR); else --- 756,761 ---- *************** *** 828,833 **** pqInternalNotice(const PGNoticeHooks *hooks, const char *fmt,...) --- 827,921 ---- } /* + * PQsetRowProcessor + * Set function that copies column data out from network buffer. + */ + void + PQsetRowProcessor(PGconn *conn, PQrowProcessor func, void *param) + { + if (!conn) + return; + + if (func) + { + /* set custom row processor */ + conn->rowProcessor = func; + conn->rowProcessorParam = param; + } + else + { + /* set default row processor */ + conn->rowProcessor = pqAddRow; + conn->rowProcessorParam = conn; + } + } + + /* + * PQgetRowProcessor + * Get current row processor of conn. set pointer to current parameter for + * row processor to param if not NULL. + */ + PQrowProcessor + PQgetRowProcessor(PGconn *conn, void **param) + { + if (!conn) + return NULL; + + if (param) + *param = conn->rowProcessorParam; + + return conn->rowProcessor; + } + + /* + * pqAddRow + * add a row to the PGresult structure, growing it if necessary + * Returns 1 if OK, -1 if error occurred. + */ + static int + pqAddRow(PGresult *res, PGrowValue *columns, void *param) + { + PGconn *conn = param; + PGresAttValue *tup; + int nfields = res->numAttributes; + int i; + + tup = (PGresAttValue *) + pqResultAlloc(res, nfields * sizeof(PGresAttValue), TRUE); + if (tup == NULL) + goto no_memory; + + for (i = 0 ; i < nfields ; i++) + { + tup[i].len = columns[i].len; + if (tup[i].len == NULL_LEN) + { + tup[i].value = res->null_field; + } + else + { + bool isbinary = (res->attDescs[i].format != 0); + tup[i].value = (char *)pqResultAlloc(res, tup[i].len + 1, isbinary); + if (tup[i].value == NULL) + goto no_memory; + + /* copy and zero-terminate */ + memcpy(tup[i].value, columns[i].value, tup[i].len); + tup[i].value[tup[i].len] = '\0'; + } + } + + if (pqAddTuple(res, tup)) + return 1; + + no_memory: + printfPQExpBuffer(&conn->errorMessage, + libpq_gettext("out of memory for query result\n")); + pqSaveErrorResult(conn); + return -1; + } + + /* * pqAddTuple * add a row pointer to the PGresult structure, growing it if necessary * Returns TRUE if OK, FALSE if not enough memory to add the row *************** *** 1223,1229 **** PQsendQueryStart(PGconn *conn) /* initialize async result-accumulation state */ conn->result = NULL; - conn->curTuple = NULL; /* ready to send command message */ return true; --- 1311,1316 ---- *************** *** 1831,1836 **** PQexecFinish(PGconn *conn) --- 1918,1975 ---- return lastResult; } + + /* + * Do-nothing row processor for PQskipResult + */ + static int + dummyRowProcessor(PGresult *res, PGrowValue *columns, void *param) + { + return 1; + } + + /* + * Exhaust remaining Data Rows in current connection. + * + * Exhaust only one resultset if skipAll is false and all + * succeeding results if true. + */ + int + PQskipResult(PGconn *conn, int skipAll) + { + PQrowProcessor savedRowProcessor; + void * savedRowProcParam; + PGresult *res; + int ret = 0; + + if (!conn) + return 0; + + /* save the current row processor settings and set dummy processor */ + savedRowProcessor = PQgetRowProcessor(conn, &savedRowProcParam); + PQsetRowProcessor(conn, dummyRowProcessor, NULL); + + /* + * Throw away the remaining rows in current result, or all succeeding + * results if skipAll is not FALSE. + */ + if (skipAll) + { + while ((res = PQgetResult(conn)) != NULL) + PQclear(res); + } + else if ((res = PQgetResult(conn)) != NULL) + { + PQclear(res); + ret = 1; + } + + PQsetRowProcessor(conn, savedRowProcessor, savedRowProcParam); + + return ret; + } + + /* * PQdescribePrepared * Obtain information about a previously prepared statement *** a/src/interfaces/libpq/fe-misc.c --- b/src/interfaces/libpq/fe-misc.c *************** *** 219,224 **** pqGetnchar(char *s, size_t len, PGconn *conn) --- 219,243 ---- } /* + * pqGetnchar: + * skip len bytes in input buffer. + */ + int + pqSkipnchar(size_t len, PGconn *conn) + { + if (len > (size_t) (conn->inEnd - conn->inCursor)) + return EOF; + + conn->inCursor += len; + + if (conn->Pfdebug) + fprintf(conn->Pfdebug, "From backend (%lu skipped)\n", + (unsigned long) len); + + return 0; + } + + /* * pqPutnchar: * write exactly len bytes to the current message */ *** a/src/interfaces/libpq/fe-protocol2.c --- b/src/interfaces/libpq/fe-protocol2.c *************** *** 569,574 **** pqParseInput2(PGconn *conn) --- 569,576 ---- /* Read another tuple of a normal query response */ if (getAnotherTuple(conn, FALSE)) return; + /* getAnotherTuple moves inStart itself */ + continue; } else { *************** *** 585,590 **** pqParseInput2(PGconn *conn) --- 587,594 ---- /* Read another tuple of a normal query response */ if (getAnotherTuple(conn, TRUE)) return; + /* getAnotherTuple moves inStart itself */ + continue; } else { *************** *** 703,721 **** failure: /* * parseInput subroutine to read a 'B' or 'D' (row data) message. ! * We add another tuple to the existing PGresult structure. * Returns: 0 if completed message, EOF if error or not enough data yet. * * Note that if we run out of data, we have to suspend and reprocess ! * the message after more data is received. We keep a partially constructed ! * tuple in conn->curTuple, and avoid reallocating already-allocated storage. */ static int getAnotherTuple(PGconn *conn, bool binary) { PGresult *result = conn->result; int nfields = result->numAttributes; ! PGresAttValue *tup; /* the backend sends us a bitmap of which attributes are null */ char std_bitmap[64]; /* used unless it doesn't fit */ --- 707,724 ---- /* * parseInput subroutine to read a 'B' or 'D' (row data) message. ! * It fills rowbuf with column pointers and then calls row processor. * Returns: 0 if completed message, EOF if error or not enough data yet. * * Note that if we run out of data, we have to suspend and reprocess ! * the message after more data is received. */ static int getAnotherTuple(PGconn *conn, bool binary) { PGresult *result = conn->result; int nfields = result->numAttributes; ! PGrowValue *rowbuf; /* the backend sends us a bitmap of which attributes are null */ char std_bitmap[64]; /* used unless it doesn't fit */ *************** *** 726,754 **** getAnotherTuple(PGconn *conn, bool binary) int bitmap_index; /* Its index */ int bitcnt; /* number of bits examined in current byte */ int vlen; /* length of the current field value */ ! result->binary = binary; ! ! /* Allocate tuple space if first time for this data message */ ! if (conn->curTuple == NULL) { ! conn->curTuple = (PGresAttValue *) ! pqResultAlloc(result, nfields * sizeof(PGresAttValue), TRUE); ! if (conn->curTuple == NULL) ! goto outOfMemory; ! MemSet(conn->curTuple, 0, nfields * sizeof(PGresAttValue)); ! ! /* ! * If it's binary, fix the column format indicators. We assume the ! * backend will consistently send either B or D, not a mix. ! */ ! if (binary) { ! for (i = 0; i < nfields; i++) ! result->attDescs[i].format = 1; } } - tup = conn->curTuple; /* Get the null-value bitmap */ nbytes = (nfields + BITS_PER_BYTE - 1) / BITS_PER_BYTE; --- 729,760 ---- int bitmap_index; /* Its index */ int bitcnt; /* number of bits examined in current byte */ int vlen; /* length of the current field value */ + const char *errmsg = libpq_gettext("unknown error\n"); ! /* resize row buffer if needed */ ! if (nfields > conn->rowBufLen) { ! rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue)); ! if (!rowbuf) { ! errmsg = libpq_gettext("out of memory for query result\n"); ! goto error_clearresult; } + conn->rowBuf = rowbuf; + conn->rowBufLen = nfields; + } + else + { + rowbuf = conn->rowBuf; + } + + result->binary = binary; + + if (binary) + { + for (i = 0; i < nfields; i++) + result->attDescs[i].format = 1; } /* Get the null-value bitmap */ nbytes = (nfields + BITS_PER_BYTE - 1) / BITS_PER_BYTE; *************** *** 757,767 **** getAnotherTuple(PGconn *conn, bool binary) { bitmap = (char *) malloc(nbytes); if (!bitmap) ! goto outOfMemory; } if (pqGetnchar(bitmap, nbytes, conn)) ! goto EOFexit; /* Scan the fields */ bitmap_index = 0; --- 763,777 ---- { bitmap = (char *) malloc(nbytes); if (!bitmap) ! { ! errmsg = libpq_gettext("out of memory for query result\n"); ! goto error_clearresult; ! } } if (pqGetnchar(bitmap, nbytes, conn)) ! goto error_clearresult; ! /* Scan the fields */ bitmap_index = 0; *************** *** 771,804 **** getAnotherTuple(PGconn *conn, bool binary) for (i = 0; i < nfields; i++) { if (!(bmap & 0200)) ! { ! /* if the field value is absent, make it a null string */ ! tup[i].value = result->null_field; ! tup[i].len = NULL_LEN; ! } else { - /* get the value length (the first four bytes are for length) */ - if (pqGetInt(&vlen, 4, conn)) - goto EOFexit; if (!binary) vlen = vlen - 4; if (vlen < 0) vlen = 0; - if (tup[i].value == NULL) - { - tup[i].value = (char *) pqResultAlloc(result, vlen + 1, binary); - if (tup[i].value == NULL) - goto outOfMemory; - } - tup[i].len = vlen; - /* read in the value */ - if (vlen > 0) - if (pqGetnchar((char *) (tup[i].value), vlen, conn)) - goto EOFexit; - /* we have to terminate this ourselves */ - tup[i].value[vlen] = '\0'; } /* advance the bitmap stuff */ bitcnt++; if (bitcnt == BITS_PER_BYTE) --- 781,809 ---- for (i = 0; i < nfields; i++) { if (!(bmap & 0200)) ! vlen = NULL_LEN; ! else if (pqGetInt(&vlen, 4, conn)) ! goto EOFexit; else { if (!binary) vlen = vlen - 4; if (vlen < 0) vlen = 0; } + + /* + * rowbuf[i].value always points to the next address of the + * length field even if the value is NULL, to allow safe + * size estimates and data copy. + */ + rowbuf[i].value = conn->inBuffer + conn->inCursor; + rowbuf[i].len = vlen; + + /* Skip the value */ + if (vlen > 0 && pqSkipnchar(vlen, conn)) + goto EOFexit; + /* advance the bitmap stuff */ bitcnt++; if (bitcnt == BITS_PER_BYTE) *************** *** 811,843 **** getAnotherTuple(PGconn *conn, bool binary) bmap <<= 1; } - /* Success! Store the completed tuple in the result */ - if (!pqAddTuple(result, tup)) - goto outOfMemory; - /* and reset for a new message */ - conn->curTuple = NULL; - if (bitmap != std_bitmap) free(bitmap); ! return 0; ! outOfMemory: ! /* Replace partially constructed result with an error result */ /* * we do NOT use pqSaveErrorResult() here, because of the likelihood that * there's not enough memory to concatenate messages... */ pqClearAsyncResult(conn); - printfPQExpBuffer(&conn->errorMessage, - libpq_gettext("out of memory for query result\n")); /* * XXX: if PQmakeEmptyPGresult() fails, there's probably not much we can * do to recover... */ conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR); conn->asyncStatus = PGASYNC_READY; /* Discard the failed message --- good idea? */ conn->inStart = conn->inEnd; --- 816,869 ---- bmap <<= 1; } if (bitmap != std_bitmap) free(bitmap); ! bitmap = NULL; ! /* tag the row as parsed */ ! conn->inStart = conn->inCursor; + /* Pass the completed row values to rowProcessor */ + switch (conn->rowProcessor(result, rowbuf, conn->rowProcessorParam)) + { + case 1: + /* everything is good */ + return 0; + + case 0: + /* processor requested early exit */ + return EOF; + + case -1: + /* if row processor already set error, then simply exit */ + if (conn->result->resultStatus != PGRES_TUPLES_OK) + return EOF; + errmsg = libpq_gettext("error in row processor\n"); + break; + + default: + /* Illegal return code */ + errmsg = libpq_gettext("invalid return value from row processor\n"); + break; + } + + error_clearresult: /* * we do NOT use pqSaveErrorResult() here, because of the likelihood that * there's not enough memory to concatenate messages... */ pqClearAsyncResult(conn); + printfPQExpBuffer(&conn->errorMessage, "%s", errmsg); + /* * XXX: if PQmakeEmptyPGresult() fails, there's probably not much we can * do to recover... */ conn->result = PQmakeEmptyPGresult(conn, PGRES_FATAL_ERROR); + conn->asyncStatus = PGASYNC_READY; + /* Discard the failed message --- good idea? */ conn->inStart = conn->inEnd; *** a/src/interfaces/libpq/fe-protocol3.c --- b/src/interfaces/libpq/fe-protocol3.c *************** *** 327,332 **** pqParseInput3(PGconn *conn) --- 327,335 ---- /* Read another tuple of a normal query response */ if (getAnotherTuple(conn, msgLength)) return; + + /* getAnotherTuple() moves inStart itself */ + continue; } else if (conn->result != NULL && conn->result->resultStatus == PGRES_FATAL_ERROR) *************** *** 613,659 **** failure: /* * parseInput subroutine to read a 'D' (row data) message. ! * We add another tuple to the existing PGresult structure. ! * Returns: 0 if completed message, EOF if error or not enough data yet. ! * ! * Note that if we run out of data, we have to suspend and reprocess ! * the message after more data is received. We keep a partially constructed ! * tuple in conn->curTuple, and avoid reallocating already-allocated storage. */ static int getAnotherTuple(PGconn *conn, int msgLength) { PGresult *result = conn->result; int nfields = result->numAttributes; ! PGresAttValue *tup; int tupnfields; /* # fields from tuple */ int vlen; /* length of the current field value */ int i; ! ! /* Allocate tuple space if first time for this data message */ ! if (conn->curTuple == NULL) ! { ! conn->curTuple = (PGresAttValue *) ! pqResultAlloc(result, nfields * sizeof(PGresAttValue), TRUE); ! if (conn->curTuple == NULL) ! goto outOfMemory; ! MemSet(conn->curTuple, 0, nfields * sizeof(PGresAttValue)); ! } ! tup = conn->curTuple; /* Get the field count and make sure it's what we expect */ if (pqGetInt(&tupnfields, 2, conn)) ! return EOF; if (tupnfields != nfields) { ! /* Replace partially constructed result with an error result */ ! printfPQExpBuffer(&conn->errorMessage, ! libpq_gettext("unexpected field count in \"D\" message\n")); ! pqSaveErrorResult(conn); ! /* Discard the failed message by pretending we read it */ ! conn->inCursor = conn->inStart + 5 + msgLength; ! return 0; } /* Scan the fields */ --- 616,661 ---- /* * parseInput subroutine to read a 'D' (row data) message. ! * It fills rowbuf with column pointers and then calls row processor. ! * Returns: 0 if completed message, EOF to cancel parsing. */ static int getAnotherTuple(PGconn *conn, int msgLength) { PGresult *result = conn->result; int nfields = result->numAttributes; ! PGrowValue *rowbuf; int tupnfields; /* # fields from tuple */ int vlen; /* length of the current field value */ int i; ! const char *errmsg = libpq_gettext("unknown error\n"); /* Get the field count and make sure it's what we expect */ if (pqGetInt(&tupnfields, 2, conn)) ! { ! /* Whole the message must be loaded on the buffer here */ ! errmsg = libpq_gettext("protocol error\n"); ! goto error_and_forward; ! } if (tupnfields != nfields) { ! errmsg = libpq_gettext("unexpected field count in \"D\" message\n"); ! goto error_and_forward; ! } ! ! /* resize row buffer if needed */ ! rowbuf = conn->rowBuf; ! if (nfields > conn->rowBufLen) ! { ! rowbuf = realloc(conn->rowBuf, nfields * sizeof(PGrowValue)); ! if (!rowbuf) ! { ! errmsg = libpq_gettext("out of memory for query result\n"); ! goto error_and_forward; ! } ! conn->rowBuf = rowbuf; ! conn->rowBufLen = nfields; } /* Scan the fields */ *************** *** 661,714 **** getAnotherTuple(PGconn *conn, int msgLength) { /* get the value length */ if (pqGetInt(&vlen, 4, conn)) - return EOF; - if (vlen == -1) { ! /* null field */ ! tup[i].value = result->null_field; ! tup[i].len = NULL_LEN; ! continue; } ! if (vlen < 0) vlen = 0; - if (tup[i].value == NULL) - { - bool isbinary = (result->attDescs[i].format != 0); ! tup[i].value = (char *) pqResultAlloc(result, vlen + 1, isbinary); ! if (tup[i].value == NULL) ! goto outOfMemory; } - tup[i].len = vlen; - /* read in the value */ - if (vlen > 0) - if (pqGetnchar((char *) (tup[i].value), vlen, conn)) - return EOF; - /* we have to terminate this ourselves */ - tup[i].value[vlen] = '\0'; } ! /* Success! Store the completed tuple in the result */ ! if (!pqAddTuple(result, tup)) ! goto outOfMemory; ! /* and reset for a new message */ ! conn->curTuple = NULL; ! return 0; ! outOfMemory: /* * Replace partially constructed result with an error result. First * discard the old result to try to win back some memory. */ pqClearAsyncResult(conn); ! printfPQExpBuffer(&conn->errorMessage, ! libpq_gettext("out of memory for query result\n")); pqSaveErrorResult(conn); - - /* Discard the failed message by pretending we read it */ - conn->inCursor = conn->inStart + 5 + msgLength; return 0; } --- 663,743 ---- { /* get the value length */ if (pqGetInt(&vlen, 4, conn)) { ! /* Whole the message must be loaded on the buffer here */ ! errmsg = libpq_gettext("protocol error\n"); ! goto error_and_forward; } ! ! if (vlen == -1) ! vlen = NULL_LEN; ! else if (vlen < 0) vlen = 0; ! /* ! * rowbuf[i].value always points to the next address of the ! * length field even if the value is NULL, to allow safe ! * size estimates and data copy. ! */ ! rowbuf[i].value = conn->inBuffer + conn->inCursor; ! rowbuf[i].len = vlen; ! ! /* Skip to the next length field */ ! if (vlen > 0 && pqSkipnchar(vlen, conn)) ! { ! /* Whole the message must be loaded on the buffer here */ ! errmsg = libpq_gettext("protocol error\n"); ! goto error_and_forward; } } ! /* tag the row as parsed, check if correctly */ ! conn->inStart += 5 + msgLength; ! if (conn->inCursor != conn->inStart) ! { ! errmsg = libpq_gettext("invalid row contents\n"); ! goto error_clearresult; ! } ! /* Pass the completed row values to rowProcessor */ ! switch (conn->rowProcessor(result, rowbuf, conn->rowProcessorParam)) ! { ! case 1: ! /* everything is good */ ! return 0; ! ! case 0: ! /* processor requested early exit - stop parsing without error*/ ! return EOF; ! ! case -1: ! /* if row processor already set error, then simply exit */ ! if (conn->result->resultStatus != PGRES_TUPLES_OK) ! return EOF; ! /* set generic row processor error */ ! errmsg = libpq_gettext("error in row processor\n"); ! goto error_clearresult; + default: + /* Illegal return code */ + errmsg = libpq_gettext("invalid return value from row processor\n"); + goto error_clearresult; + } + + + error_and_forward: + /* Discard the failed message by pretending we read it */ + conn->inCursor = conn->inStart + 5 + msgLength; + + error_clearresult: /* * Replace partially constructed result with an error result. First * discard the old result to try to win back some memory. */ pqClearAsyncResult(conn); ! printfPQExpBuffer(&conn->errorMessage, "%s", errmsg); pqSaveErrorResult(conn); return 0; } *** a/src/interfaces/libpq/libpq-fe.h --- b/src/interfaces/libpq/libpq-fe.h *************** *** 149,154 **** typedef struct pgNotify --- 149,165 ---- struct pgNotify *next; /* list link */ } PGnotify; + /* PGrowValue points a column value of in network buffer. + * Value is a string without null termination and length len. + * NULL is represented as len < 0, value points then to place + * where value would have been. + */ + typedef struct pgRowValue + { + int len; /* length in bytes of the value */ + char *value; /* actual value, without null termination */ + } PGrowValue; + /* Function types for notice-handling callbacks */ typedef void (*PQnoticeReceiver) (void *arg, const PGresult *res); typedef void (*PQnoticeProcessor) (void *arg, const char *message); *************** *** 416,421 **** extern PGPing PQping(const char *conninfo); --- 427,464 ---- extern PGPing PQpingParams(const char *const * keywords, const char *const * values, int expand_dbname); + /* + * Typedef for alternative row processor. + * + * Columns array will contain PQnfields() entries, each one pointing + * to particular column data in network buffer. This function is + * supposed to copy data out from there and store somewhere. NULL is + * signified with len<0. + * + * This function must return 1 for success and -1 for failure and the + * caller relreases the current PGresult for the case. Returning 0 + * instructs libpq to exit immediately from the topmost libpq function + * without releasing PGresult under work. + */ + typedef int (*PQrowProcessor)(PGresult *res, PGrowValue *columns, + void *param); + + /* + * Set alternative row data processor for PGconn. + * + * By registering this function, pg_result disables its own result + * store and calls it for rows one by one. + * + * func is row processor function. See the typedef RowProcessor. + * + * rowProcessorParam is the contextual variable that passed to + * RowProcessor. + */ + extern void PQsetRowProcessor(PGconn *conn, PQrowProcessor func, + void *rowProcessorParam); + extern PQrowProcessor PQgetRowProcessor(PGconn *conn, void **param); + extern int PQskipResult(PGconn *conn, int skipAll); + /* Force the write buffer to be written (or at least try) */ extern int PQflush(PGconn *conn); *** a/src/interfaces/libpq/libpq-int.h --- b/src/interfaces/libpq/libpq-int.h *************** *** 398,404 **** struct pg_conn /* Status for asynchronous result construction */ PGresult *result; /* result being constructed */ - PGresAttValue *curTuple; /* tuple currently being read */ #ifdef USE_SSL bool allow_ssl_try; /* Allowed to try SSL negotiation */ --- 398,403 ---- *************** *** 441,446 **** struct pg_conn --- 440,453 ---- /* Buffer for receiving various parts of messages */ PQExpBufferData workBuffer; /* expansible string */ + + /* + * Read column data from network buffer. + */ + PQrowProcessor rowProcessor;/* Function pointer */ + void *rowProcessorParam; /* Contextual parameter for rowProcessor */ + PGrowValue *rowBuf; /* Buffer for passing values to rowProcessor */ + int rowBufLen; /* Number of columns allocated in rowBuf */ }; /* PGcancel stores all data necessary to cancel a connection. A copy of this *************** *** 558,563 **** extern int pqGets(PQExpBuffer buf, PGconn *conn); --- 565,571 ---- extern int pqGets_append(PQExpBuffer buf, PGconn *conn); extern int pqPuts(const char *s, PGconn *conn); extern int pqGetnchar(char *s, size_t len, PGconn *conn); + extern int pqSkipnchar(size_t len, PGconn *conn); extern int pqPutnchar(const char *s, size_t len, PGconn *conn); extern int pqGetInt(int *result, size_t bytes, PGconn *conn); extern int pqPutInt(int value, size_t bytes, PGconn *conn);