connection.c

来自「postgresql-odbc,跨平台应用」· C语言 代码 · 共 2,634 行 · 第 1/5 页

			CONNLOCK_RELEASE(conn);
			SOCK_Destructor(conn->sock);
			CONNLOCK_ACQUIRE(conn);
			conn->sock = NULL;
		}
	}
	else if (set_no_trans)
	{
		CONNLOCK_RELEASE(conn);
		CC_discard_marked_objects(conn);
		CONNLOCK_ACQUIRE(conn);
	}
	if (conn->result_uncommitted)
	{
		CONNLOCK_RELEASE(conn);
		ProcessRollback(conn, TRUE, FALSE);
		CONNLOCK_ACQUIRE(conn);
		conn->result_uncommitted = 0;
	}
	CONNLOCK_RELEASE(conn);
}

void	CC_on_abort_partial(ConnectionClass *conn)
{
mylog("CC_on_abort_partial in\n");
	ProcessRollback(conn, TRUE, TRUE);
	CONNLOCK_ACQUIRE(conn);
	CC_discard_marked_objects(conn);
	CONNLOCK_RELEASE(conn);
}

static BOOL
is_setting_search_path(const UCHAR* query)
{
	for (query += 4; *query; query++)
	{
		if (!isspace(*query))
		{
			if (strnicmp(query, "search_path", 11) == 0)
				return TRUE;
			query++;
			while (*query && !isspace(*query))
				query++;
		}
	}
	return FALSE;
}

/*
 *	The "result_in" is only used by QR_next_tuple() to fetch another group of rows into
 *	the same existing QResultClass (this occurs when the tuple cache is depleted and
 *	needs to be re-filled).
 *
 *	The "cursor" is used by SQLExecute to associate a statement handle as the cursor name
 *	(i.e., C3326857) for SQL select statements.  This cursor is then used in future
 *	'declare cursor C3326857 for ...' and 'fetch 100 in C3326857' statements.
 */
QResultClass *
CC_send_query_append(ConnectionClass *self, const char *query, QueryInfo *qi, UDWORD flag, StatementClass *stmt, const char *appendq)
{
	CSTR	func = "CC_send_query";
	QResultClass *cmdres = NULL,
			   *retres = NULL,
			   *res = NULL;
	BOOL	ignore_abort_on_conn = ((flag & IGNORE_ABORT_ON_CONN) != 0),
		create_keyset = ((flag & CREATE_KEYSET) != 0),
		issue_begin = ((flag & GO_INTO_TRANSACTION) != 0 && !CC_is_in_trans(self)),
		rollback_on_error, query_rollback, end_with_commit;

	const char	*wq;
	char		swallow, *ptr;
	CSTR	svpcmd = "SAVEPOINT";
	CSTR	per_query_svp = "_per_query_svp_";
	CSTR	rlscmd = "RELEASE";
	static size_t	lenbgncmd = 0, lenrbkcmd = 0, lensvpcmd = 0,
			lenrlscmd = 0, lenperqsvp = 0;
	size_t	qrylen;
	int			id;
	SocketClass *sock = self->sock;
	int			maxlen,
				empty_reqs;
	BOOL		msg_truncated,
				ReadyToReturn = FALSE,
				query_completed = FALSE,
				beforeV2 = PG_VERSION_LT(self, 6.4),
				aborted = FALSE,
				used_passed_result_object = FALSE,
			discard_next_begin = FALSE,
			discard_next_savepoint = FALSE,
			consider_rollback;
	Int4		response_length;
	UInt4		leng;
	ConnInfo	*ci = &(self->connInfo);
	int		func_cs_count = 0;

	/* ERROR_MSG_LENGTH is suffcient */
	char msgbuffer[ERROR_MSG_LENGTH + 1];

	/* QR_set_command() dups this string so doesn't need static */
	char		cmdbuffer[ERROR_MSG_LENGTH + 1];

	if (appendq)
	{
		mylog("%s_append: conn=%p, query='%s'+'%s'\n", func, self, query, appendq);
		qlog("conn=%p, query='%s'+'%s'\n", self, query, appendq);
	}
	else
	{
		mylog("%s: conn=%p, query='%s'\n", func, self, query);
		qlog("conn=%p, query='%s'\n", self, query);
	}

	if (!self->sock)
	{
		CC_set_error(self, CONNECTION_COULD_NOT_SEND, "Could not send Query(connection dead)", func);
		CC_on_abort(self, CONN_DEAD);
		return NULL;
	}

	/* Indicate that we are sending a query to the backend */
	maxlen = CC_get_max_query_len(self);
	qrylen = strlen(query);
	if (maxlen > 0 && maxlen < (int) qrylen + 1)
	{
		CC_set_error(self, CONNECTION_MSG_TOO_LONG, "Query string is too long", func);
		return NULL;
	}

	if ((NULL == query) || (query[0] == '\0'))
		return NULL;

	if (SOCK_get_errcode(sock) != 0)
	{
		CC_set_error(self, CONNECTION_COULD_NOT_SEND, "Could not send Query to backend", func);
		CC_on_abort(self, CONN_DEAD);
		return NULL;
	}

	rollback_on_error = (flag & ROLLBACK_ON_ERROR) != 0;
	end_with_commit = (flag & END_WITH_COMMIT) != 0;
#define	return DONT_CALL_RETURN_FROM_HERE???
	ENTER_INNER_CONN_CS(self, func_cs_count);
	consider_rollback = (issue_begin || (CC_is_in_trans(self) && !CC_is_in_error_trans(self)) || strnicmp(query, "begin", 5) == 0);
	if (rollback_on_error)
		rollback_on_error = consider_rollback;
	query_rollback = (rollback_on_error && !end_with_commit && PG_VERSION_GE(self, 8.0));
	if (!query_rollback && consider_rollback && !end_with_commit)
	{
		if (stmt)
		{
			StatementClass	*astmt = SC_get_ancestor(stmt);
			if (!SC_accessed_db(astmt))
			{
				if (SQL_ERROR == SetStatementSvp(astmt))
				{
					SC_set_error(stmt, STMT_INTERNAL_ERROR, "internal savepoint error", func);
					goto cleanup;
				}
			}
		}
	}

	SOCK_put_char(sock, 'Q');
	if (SOCK_get_errcode(sock) != 0)
	{
		CC_set_error(self, CONNECTION_COULD_NOT_SEND, "Could not send Query to backend", func);
		CC_on_abort(self, CONN_DEAD);
		goto cleanup;
	}
	if (stmt)
		SC_forget_unnamed(stmt);

	if (!lenbgncmd)
	{
		lenbgncmd = strlen(bgncmd);
		lensvpcmd = strlen(svpcmd);
		lenrbkcmd = strlen(rbkcmd);
		lenrlscmd = strlen(rlscmd);
		lenperqsvp = strlen(per_query_svp);
	}
	if (PROTOCOL_74(ci))
	{
		leng = (UInt4) qrylen;
		if (appendq)
			leng += (UInt4) (strlen(appendq) + 1);
		if (issue_begin)
			leng += (UInt4) (lenbgncmd + 1);
		if (query_rollback)
		{
			leng += (UInt4) (lensvpcmd + 1 + lenperqsvp + 1);
			leng += (UInt4) (1 + lenrlscmd + 1 + lenperqsvp);
		}
		leng++;
		SOCK_put_int(sock, leng + 4, 4);
inolog("leng=%d\n", leng);
	}
	if (issue_begin)
	{
		SOCK_put_n_char(sock, bgncmd, lenbgncmd);
		SOCK_put_n_char(sock, semi_colon, 1);
		discard_next_begin = TRUE;
	}
	if (query_rollback)
	{
		char cmd[64];

		snprintf(cmd, sizeof(cmd), "%s %s;", svpcmd, per_query_svp);
		SOCK_put_n_char(sock, cmd, (Int4) strlen(cmd));
		discard_next_savepoint = TRUE;
	}
	SOCK_put_n_char(sock, query, qrylen);
	if (appendq)
	{
		SOCK_put_n_char(sock, semi_colon, 1);
		SOCK_put_n_char(sock, appendq, strlen(appendq));
	}
	if (query_rollback)
	{
		char cmd[64];

		snprintf(cmd, sizeof(cmd), ";%s %s", rlscmd, per_query_svp);
		SOCK_put_n_char(sock, cmd, (Int4) strlen(cmd));
	}
	SOCK_put_n_char(sock, NULL_STRING, 1);
	leng = SOCK_flush_output(sock);

	if (SOCK_get_errcode(sock) != 0)
	{
		CC_set_error(self, CONNECTION_COULD_NOT_SEND, "Could not send Query to backend", func);
		CC_on_abort(self, CONN_DEAD);
		goto cleanup;
	}

	mylog("send_query: done sending query %dbytes flushed\n", leng);
 
	empty_reqs = 0;
	for (wq = query; isspace((UCHAR) *wq); wq++)
		;
	if (*wq == '\0')
		empty_reqs = 1;
	cmdres = qi ? qi->result_in : NULL;
	if (cmdres)
		used_passed_result_object = TRUE;
	else
	{
		cmdres = QR_Constructor();
		if (!cmdres)
		{
			CC_set_error(self, CONNECTION_COULD_NOT_RECEIVE, "Could not create result info in send_query.", func);
			goto cleanup;
		}
	}
	res = cmdres;
	while (!ReadyToReturn)
	{
		/* what type of message is coming now ? */
		id = SOCK_get_id(sock);

		if ((SOCK_get_errcode(sock) != 0) || (id == EOF))
		{
			CC_set_error(self, CONNECTION_NO_RESPONSE, "No response from the backend", func);

			mylog("send_query: 'id' - %s\n", CC_get_errormsg(self));
			CC_on_abort(self, CONN_DEAD);
			ReadyToReturn = TRUE;
			retres = NULL;
			break;
		}

		mylog("send_query: got id = '%c'\n", id);

		response_length = SOCK_get_response_length(sock);
inolog("send_query response_length=%d\n", response_length);
		switch (id)
		{
			case 'A':			/* Asynchronous Messages are ignored */
				(void) SOCK_get_int(sock, 4);	/* id of notification */
				SOCK_get_string(sock, msgbuffer, ERROR_MSG_LENGTH);
				/* name of the relation the message comes from */
				break;
			case 'C':			/* portal query command, no tuples
								 * returned */
				/* read in the return message from the backend */
				SOCK_get_string(sock, cmdbuffer, ERROR_MSG_LENGTH);
				if (SOCK_get_errcode(sock) != 0)
				{
					CC_set_error(self, CONNECTION_NO_RESPONSE, "No response from backend while receiving a portal query command", func);
					mylog("send_query: 'C' - %s\n", CC_get_errormsg(self));
					CC_on_abort(self, CONN_DEAD);
					ReadyToReturn = TRUE;
					retres = NULL;
				}
				else
				{
					mylog("send_query: ok - 'C' - %s\n", cmdbuffer);

					if (query_completed)	/* allow for "show" style notices */
					{
						res->next = QR_Constructor();
						res = res->next;
					} 

					mylog("send_query: setting cmdbuffer = '%s'\n", cmdbuffer);

					trim(cmdbuffer); /* get rid of trailing space */ 
					if (strnicmp(cmdbuffer, bgncmd, lenbgncmd) == 0)
					{
						CC_set_in_trans(self);
						if (discard_next_begin) /* disicard the automatically issued BEGIN */
						{
							discard_next_begin = FALSE;
							continue; /* discard the result */
						}
					}
					else if (strnicmp(cmdbuffer, svpcmd, lensvpcmd) == 0)
					{
						if (discard_next_savepoint)
						{
inolog("Discarded the first SAVEPOINT\n");
							discard_next_savepoint = FALSE;
							continue; /* discard the result */
						}
					}
					else if (strnicmp(cmdbuffer, rbkcmd, lenrbkcmd) == 0)
					{
						if (PROTOCOL_74(&(self->connInfo)))
							CC_set_in_error_trans(self); /* mark the transaction error in case of manual rollback */
						else
							CC_on_abort(self, NO_TRANS);
					}
					else
					{
						ptr = strrchr(cmdbuffer, ' ');
						if (ptr)
							res->recent_processed_row_count = atoi(ptr + 1);
						else
							res->recent_processed_row_count = -1;
						if (PROTOCOL_74(&(self->connInfo)))
						{
							if (NULL != self->current_schema &&
							    strnicmp(cmdbuffer, "SET", 3) == 0)
							{
								if (is_setting_search_path(query))
									reset_current_schema(self);
							}
						}
						else
						{
							if (strnicmp(cmdbuffer, "COMMIT", 6) == 0)
								CC_on_commit(self);
							else if (strnicmp(cmdbuffer, "END", 3) == 0)
								CC_on_commit(self);
							else if (strnicmp(cmdbuffer, "ABORT", 5) == 0)
								CC_on_abort(self, NO_TRANS);
						}
					}

					if (QR_command_successful(res))
						QR_set_rstatus(res, PORES_COMMAND_OK);
					QR_set_command(res, cmdbuffer);
					query_completed = TRUE;
					mylog("send_query: returning res = %p\n", res);
					if (!beforeV2)
						break;

					/*
					 * (Quotation from the original comments) since
					 * backend may produce more than one result for some
					 * commands we need to poll until clear so we send an
					 * empty query, and keep reading out of the pipe until
					 * an 'I' is received
					 */

					if (empty_reqs == 0)
					{
						SOCK_put_string(sock, "Q ");
						SOCK_flush_output(sock);
						empty_reqs++;
					}
				}
				break;
			case 'Z':			/* Backend is ready for new query (6.4) */
				if (empty_reqs == 0)
				{
					ReadyToReturn = TRUE;
					if (aborted || query_completed)
						retres = cmdres;
					else
						ReadyToReturn = FALSE;
				}
				EatReadyForQuery(self);
				break;
			case 'N':			/* NOTICE: */
				msg_truncated = handle_notice_message(self, cmdbuffer, sizeof(cmdbuffer), res->sqlstate, "send_query", res);
				break;		/* dont return a result -- continue
								 * reading */

			case 'I':			/* The server sends an empty query */
				/* There is a closing '\0' following the 'I', so we eat it */
				if (PROTOCOL_74(ci) && 0 == response_length)
					swallow = '\0';
				else
					swallow = SOCK_get_char(sock);
				if ((swallow != '\0') || SOCK_get_errcode(sock) != 0)
				{
					CC_set_errornumber(self, CONNECTION_BACKEND_CRAZY);
					QR_set_message(res, "Unexpected protocol character from backend (send_query - I)");
					QR_set_rstatus(res, PORES_FATAL_ERROR);
					ReadyToReturn = TRUE;
					retres = cmdres;
					break;
				}
				else
				{
					/* We return the empty query */
					QR_set_rstatus(res, PORES_EMPTY_QUERY);
				}
				if (empty_reqs > 0)
				{
					if (--empty_reqs == 0)
						query_completed = TRUE;
				}
				break;
			case 'E':
				msg_truncated = handle_error_message(self, msgbuffer, sizeof(msgbuffer), res->sqlstate, "send_query", res);

				/* We should report that an error occured. Zoltan */
				aborted = TRUE;

				query_completed = TRUE;
				break;

			case 'P':			/* get the Portal name */
				SOCK_get_string(sock, msgbuffer, ERROR_MSG_LENGTH);
				break;
			case 'T':			/* Tuple results start here */
				if (query_completed)
				{
					res->next = QR_Constructor();
					if (!res->next)
					{
						CC_set_error(self, CONNECTION_COULD_NOT_RECEIVE, "Could not create result info in send_query.", func);
						ReadyToReturn = TRUE;
						retres = NULL;
						break;
					}
					if (create_keyset)
					{
						QR_set_haskeyset(res->next);
						if (stmt)
							res->num_key_fields = stmt->num_key_fields;
					}
					mylog("send_query: 'T' no result_in: res = %p\n", res->next);
					res = res->next;

					if (qi)
						QR_set_cache_size(res, qi->row_size);
				}
				if (!used_passed_result_object)
				{
					const char *cursor = qi ? qi->cursor : NULL;
					if (create_keyset)
					{
						QR_set_haskeyset(res);
						if (stmt)
							res->num_key_fields = stmt->num_key_fields;
						if (cursor && cursor[0])
							QR_set_synchronize_keys(res);
					}
					if (!QR_fetch_tuples(res, self, cursor))
					{
						CC_set_error(self, CONNECTION_COULD_NOT_RECEIVE, QR_get_message(res), func);
						ReadyToReturn = TRUE;
						if (PORES_FATAL_ERROR == QR_get_rstatus(res))
							retres = cmdres;
						else
							retres = NULL;
						break;
					}
					query_completed = TRUE;
				}
				else
				{				/* next fetch, so reuse an existing result */

					/*
					 * called from QR_next_tuple and must return
					 * immediately.
					 */
					ReadyToReturn = TRUE;
					if (!QR_fetch_tuples(res, NULL, NULL))
					{
						CC_set_error(self, CONNECTION_COULD_NOT_RECEIVE, QR_get_message(res), func);
						retres = NULL;
						break;
					}
					retres = cmdres;
				}
				break;
			case 'D':			/* Copy in command began successfully */
				if (query_completed)
				{
					res->next = QR_Constructor();
					res = res->next;
				}
				QR_set_rstatus(res, PORES_COPY_IN);
				ReadyToReturn = TRUE;
				retres = cmdres;
				break;
			case 'B':			/* Copy out command began successfully */
				if (query_completed)
				{
					res->next = QR_Constructor();
					res = res->next;
				}
				QR_set_rstatus(res, PORES_COPY_OUT);
				ReadyToReturn = TRUE;
				retres = cmdres;
				break;
			case 'S':		/* parameter status */
				getParameterValues(self);
				break;
			case 's':		/* portal suspended */
				QR_set_no_fetching_tuples(res);
				break;
			default:
				/* skip the unexpected response i