fe-protocol3.c

PostgreSQL 8.1.4的源码 适用于Linux下的开源数据库系统

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 */
	for (i = 0; i < nfields; i++)
	{
		/* 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;
}


/*
 * Attempt to read an Error or Notice response message.
 * This is possible in several places, so we break it out as a subroutine.
 * Entry: 'E' or 'N' message type and length have already been consumed.
 * Exit: returns 0 if successfully consumed message.
 *		 returns EOF if not enough data.
 */
int
pqGetErrorNotice3(PGconn *conn, bool isError)
{
	PGresult   *res = NULL;
	PQExpBufferData workBuf;
	char		id;
	const char *val;

	/*
	 * Since the fields might be pretty long, we create a temporary
	 * PQExpBuffer rather than using conn->workBuffer.	workBuffer is intended
	 * for stuff that is expected to be short.	We shouldn't use
	 * conn->errorMessage either, since this might be only a notice.
	 */
	initPQExpBuffer(&workBuf);

	/*
	 * Make a PGresult to hold the accumulated fields.	We temporarily lie
	 * about the result status, so that PQmakeEmptyPGresult doesn't uselessly
	 * copy conn->errorMessage.
	 */
	res = PQmakeEmptyPGresult(conn, PGRES_EMPTY_QUERY);
	if (!res)
		goto fail;
	res->resultStatus = isError ? PGRES_FATAL_ERROR : PGRES_NONFATAL_ERROR;

	/*
	 * Read the fields and save into res.
	 */
	for (;;)
	{
		if (pqGetc(&id, conn))
			goto fail;
		if (id == '\0')
			break;				/* terminator found */
		if (pqGets(&workBuf, conn))
			goto fail;
		pqSaveMessageField(res, id, workBuf.data);
	}

	/*
	 * Now build the "overall" error message for PQresultErrorMessage.
	 */
	resetPQExpBuffer(&workBuf);
	val = PQresultErrorField(res, PG_DIAG_SEVERITY);
	if (val)
		appendPQExpBuffer(&workBuf, "%s:  ", val);
	if (conn->verbosity == PQERRORS_VERBOSE)
	{
		val = PQresultErrorField(res, PG_DIAG_SQLSTATE);
		if (val)
			appendPQExpBuffer(&workBuf, "%s: ", val);
	}
	val = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY);
	if (val)
		appendPQExpBufferStr(&workBuf, val);
	val = PQresultErrorField(res, PG_DIAG_STATEMENT_POSITION);
	if (val)
	{
		/* translator: %s represents a digit string */
		appendPQExpBuffer(&workBuf, libpq_gettext(" at character %s"), val);
	}
	else
	{
		val = PQresultErrorField(res, PG_DIAG_INTERNAL_POSITION);
		if (val)
		{
			/* translator: %s represents a digit string */
			appendPQExpBuffer(&workBuf, libpq_gettext(" at character %s"),
							  val);
		}
	}
	appendPQExpBufferChar(&workBuf, '\n');
	if (conn->verbosity != PQERRORS_TERSE)
	{
		val = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL);
		if (val)
			appendPQExpBuffer(&workBuf, libpq_gettext("DETAIL:  %s\n"), val);
		val = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT);
		if (val)
			appendPQExpBuffer(&workBuf, libpq_gettext("HINT:  %s\n"), val);
		val = PQresultErrorField(res, PG_DIAG_INTERNAL_QUERY);
		if (val)
			appendPQExpBuffer(&workBuf, libpq_gettext("QUERY:  %s\n"), val);
		val = PQresultErrorField(res, PG_DIAG_CONTEXT);
		if (val)
			appendPQExpBuffer(&workBuf, libpq_gettext("CONTEXT:  %s\n"), val);
	}
	if (conn->verbosity == PQERRORS_VERBOSE)
	{
		const char *valf;
		const char *vall;

		valf = PQresultErrorField(res, PG_DIAG_SOURCE_FILE);
		vall = PQresultErrorField(res, PG_DIAG_SOURCE_LINE);
		val = PQresultErrorField(res, PG_DIAG_SOURCE_FUNCTION);
		if (val || valf || vall)
		{
			appendPQExpBufferStr(&workBuf, libpq_gettext("LOCATION:  "));
			if (val)
				appendPQExpBuffer(&workBuf, libpq_gettext("%s, "), val);
			if (valf && vall)	/* unlikely we'd have just one */
				appendPQExpBuffer(&workBuf, libpq_gettext("%s:%s"),
								  valf, vall);
			appendPQExpBufferChar(&workBuf, '\n');
		}
	}

	/*
	 * Either save error as current async result, or just emit the notice.
	 */
	if (isError)
	{
		res->errMsg = pqResultStrdup(res, workBuf.data);
		if (!res->errMsg)
			goto fail;
		pqClearAsyncResult(conn);
		conn->result = res;
		resetPQExpBuffer(&conn->errorMessage);
		appendPQExpBufferStr(&conn->errorMessage, workBuf.data);
	}
	else
	{
		/* We can cheat a little here and not copy the message. */
		res->errMsg = workBuf.data;
		if (res->noticeHooks.noticeRec != NULL)
			(*res->noticeHooks.noticeRec) (res->noticeHooks.noticeRecArg, res);
		PQclear(res);
	}

	termPQExpBuffer(&workBuf);
	return 0;

fail:
	PQclear(res);
	termPQExpBuffer(&workBuf);
	return EOF;
}


/*
 * Attempt to read a ParameterStatus message.
 * This is possible in several places, so we break it out as a subroutine.
 * Entry: 'S' message type and length have already been consumed.
 * Exit: returns 0 if successfully consumed message.
 *		 returns EOF if not enough data.
 */
static int
getParameterStatus(PGconn *conn)
{
	PQExpBufferData valueBuf;

	/* Get the parameter name */
	if (pqGets(&conn->workBuffer, conn))
		return EOF;
	/* Get the parameter value (could be large) */
	initPQExpBuffer(&valueBuf);
	if (pqGets(&valueBuf, conn))
	{
		termPQExpBuffer(&valueBuf);
		return EOF;
	}
	/* And save it */
	pqSaveParameterStatus(conn, conn->workBuffer.data, valueBuf.data);
	termPQExpBuffer(&valueBuf);
	return 0;
}


/*
 * Attempt to read a Notify response message.
 * This is possible in several places, so we break it out as a subroutine.
 * Entry: 'A' message type and length have already been consumed.
 * Exit: returns 0 if successfully consumed Notify message.
 *		 returns EOF if not enough data.
 */
static int
getNotify(PGconn *conn)
{
	int			be_pid;
	char	   *svname;
	int			nmlen;
	int			extralen;
	PGnotify   *newNotify;

	if (pqGetInt(&be_pid, 4, conn))
		return EOF;
	if (pqGets(&conn->workBuffer, conn))
		return EOF;
	/* must save name while getting extra string */
	svname = strdup(conn->workBuffer.data);
	if (!svname)
		return EOF;
	if (pqGets(&conn->workBuffer, conn))
	{
		free(svname);
		return EOF;
	}

	/*
	 * Store the strings right after the PQnotify structure so it can all be
	 * freed at once.  We don't use NAMEDATALEN because we don't want to tie
	 * this interface to a specific server name length.
	 */
	nmlen = strlen(svname);
	extralen = strlen(conn->workBuffer.data);
	newNotify = (PGnotify *) malloc(sizeof(PGnotify) + nmlen + extralen + 2);
	if (newNotify)
	{
		newNotify->relname = (char *) newNotify + sizeof(PGnotify);
		strcpy(newNotify->relname, svname);
		newNotify->extra = newNotify->relname + nmlen + 1;
		strcpy(newNotify->extra, conn->workBuffer.data);
		newNotify->be_pid = be_pid;
		newNotify->next = NULL;
		if (conn->notifyTail)
			conn->notifyTail->next = newNotify;
		else
			conn->notifyHead = newNotify;
		conn->notifyTail = newNotify;
	}

	free(svname);
	return 0;
}

/*
 * getCopyStart - process CopyInResponse or CopyOutResponse message
 *
 * parseInput already read the message type and length.
 */
static int
getCopyStart(PGconn *conn, ExecStatusType copytype)
{
	PGresult   *result;
	int			nfields;
	int			i;

	result = PQmakeEmptyPGresult(conn, copytype);
	if (!result)
		goto failure;

	if (pqGetc(&conn->copy_is_binary, conn))
		goto failure;
	result->binary = conn->copy_is_binary;
	/* the next two bytes are the number of fields	*/
	if (pqGetInt(&(result->numAttributes), 2, conn))
		goto failure;
	nfields = result->numAttributes;

	/* allocate space for the attribute descriptors */
	if (nfields > 0)
	{
		result->attDescs = (PGresAttDesc *)
			pqResultAlloc(result, nfields * sizeof(PGresAttDesc), TRUE);
		if (!result->attDescs)
			goto failure;
		MemSet(result->attDescs, 0, nfields * sizeof(PGresAttDesc));
	}

	for (i = 0; i < nfields; i++)
	{
		int			format;

		if (pqGetInt(&format, 2, conn))
			goto failure;

		/*
		 * Since pqGetInt treats 2-byte integers as unsigned, we need to
		 * coerce these results to signed form.
		 */
		format = (int) ((int16) format);
		result->attDescs[i].format = format;
	}

	/* Success! */
	conn->result = result;
	return 0;

failure:
	PQclear(result);
	return EOF;
}

/*
 * getReadyForQuery - process ReadyForQuery message
 */
static int
getReadyForQuery(PGconn *conn)
{
	char		xact_status;

	if (pqGetc(&xact_status, conn))
		return EOF;
	switch (xact_status)
	{
		case 'I':
			conn->xactStatus = PQTRANS_IDLE;
			break;
		case 'T':
			conn->xactStatus = PQTRANS_INTRANS;
			break;
		case 'E':
			conn->xactStatus = PQTRANS_INERROR;
			break;
		default:
			conn->xactStatus = PQTRANS_UNKNOWN;
			break;
	}

	return 0;
}

/*
 * PQgetCopyData - read a row of data from the backend during COPY OUT
 *
 * If successful, sets *buffer to point to a malloc'd row of data, and
 * returns row length (always > 0) as result.
 * Returns 0 if no row available yet (only possible if async is true),
 * -1 if end of copy (consult PQgetResult), or -2 if error (consult
 * PQerrorMessage).
 */
int
pqGetCopyData3(PGconn *conn, char **buffer, int async)
{
	char		id;
	int			msgLength;
	int			avail;

	for (;;)
	{
		/*
		 * Do we have the next input message?  To make life simpler for async
		 * callers, we keep returning 0 until the next message is fully
		 * available, even if it is not Copy Data.
		 */
		conn->inCursor = conn->inStart;
		if (pqGetc(&id, conn))
			goto nodata;
		if (pqGetInt(&msgLength, 4, conn))
			goto nodata;
		avail = conn->inEnd - conn->inCursor;
		if (avail < msgLength - 4)
			goto nodata;

		/*
		 * If it's anything except Copy Data, exit COPY_OUT mode and let
		 * caller read status with PQgetResult().  The normal case is that
		 * it's Copy Done, but we let parseInput read that.
		 */
		if (id != 'd')
		{
			conn->asyncStatus = PGASYNC_BUSY;
			return -1;
		}

		/*
		 * Drop zero-length messages (shouldn't happen anyway).  Otherwise
		 * pass the data back to the caller.
		 */
		msgLength -= 4;
		if (msgLength > 0)
		{
			*buffer = (char *) malloc(msgLength + 1);
			if (*buffer == NULL)
			{
				printfPQExpBuffer(&conn->errorMessage,
								  libpq_gettext("out of memory\n"));
				return -2;
			}
			memcpy(*buffer, &conn->inBuffer[conn->inCursor], msgLength);
			(*buffer)[msgLength] = '\0';		/* Add terminating null */

			/* Mark message consumed */
			conn->inStart = conn->inCursor + msgLength;

			return msgLength;
		}

		/* Empty, so drop it and loop around for another */
		conn->inStart = conn->inCursor;
		continue;

nodata:
		/* Don't block if async read requested */
		if (async)
			return 0;
		/* Need to load more data */
		if (pqWait(TRUE, FALSE, conn) ||
			pqReadData(conn) < 0)
			return -2;
	}
}

/*
 * PQgetline - gets a newline-terminated string from the backend.
 *
 * See fe-exec.c for documentation.
 */
int
pqGetline3(PGconn *conn, char *s, int maxlen)
{
	int			status;

	if (conn->sock < 0 ||
		conn->asyncStatus != PGASYNC_COPY_OUT ||
		conn->copy_is_binary)
	{
		printfPQExpBuffer(&conn->errorMessage,