routerconnection.c

适合于Unix/Linux下的一个持久数据库连接池

						const char *value, 
						uint32_t valuesize,
						int16_t *isnull) {
	cur->inputBindBlob(variable+1,value,valuesize);
	return true;
}

bool routercursor::inputBindClob(const char *variable, 
						uint16_t variablesize,
						const char *value, 
						uint32_t valuesize,
						int16_t *isnull) {
	cur->inputBindClob(variable+1,value,valuesize);
	return true;
}

bool routercursor::outputBindString(const char *variable, 
						uint16_t variablesize,
						char *value,
						uint16_t valuesize,
						int16_t *isnull) {
	cur->defineOutputBindString(variable+1,valuesize);
	obv[obcount].variable=variable+1;
	obv[obcount].type=STRING_BIND;
	obv[obcount].value.stringvalue=value;
	obv[obcount].valuesize=valuesize;
	obv[obcount].isnull=isnull;
	obcount++;
	return true;
}

bool routercursor::outputBindInteger(const char *variable, 
						uint16_t variablesize,
						int64_t *value,
						int16_t *isnull) {
	cur->defineOutputBindInteger(variable+1);
	obv[obcount].variable=variable+1;
	obv[obcount].type=INTEGER_BIND;
	obv[obcount].value.intvalue=value;
	obv[obcount].isnull=isnull;
	obcount++;
	return true;
}

bool routercursor::outputBindDouble(const char *variable, 
						uint16_t variablesize,
						double *value,
						uint32_t *precision,
						uint32_t *scale,
						int16_t *isnull) {
	cur->defineOutputBindDouble(variable+1);
	obv[obcount].variable=variable+1;
	obv[obcount].type=DOUBLE_BIND;
	obv[obcount].value.doublevalue=value;
	obv[obcount].isnull=isnull;
	obcount++;
	return true;
}

bool routercursor::outputBindBlob(const char *variable, 
						uint16_t variablesize,
						uint16_t index,
						int16_t *isnull) {
	cur->defineOutputBindBlob(variable+1);
	obv[obcount].variable=variable+1;
	obv[obcount].type=BLOB_BIND;
	obv[obcount].isnull=isnull;
	obcount++;
	return true;
}

bool routercursor::outputBindClob(const char *variable, 
						uint16_t variablesize,
						uint16_t index,
						int16_t *isnull) {
	cur->defineOutputBindClob(variable+1);
	obv[obcount].variable=variable+1;
	obv[obcount].type=CLOB_BIND;
	obv[obcount].isnull=isnull;
	obcount++;
	return true;
}

bool routercursor::outputBindCursor(const char *variable,
						uint16_t variablesize,
						sqlrcursor_svr *cursor) {
	cur->defineOutputBindCursor(variable+1);
	cbv[cbcount].variable=variable+1;
	cbv[cbcount].cursor=cursor;
	cbcount++;
	return true;
}

void routercursor::returnOutputBindBlob(uint16_t index) {
	const char	*varname=obv[index].variable;
	uint32_t	length=cur->getOutputBindLength(varname);
	conn->startSendingLong(length);
	conn->sendLongSegment(cur->getOutputBindBlob(varname),length);
	conn->endSendingLong();
}

void routercursor::returnOutputBindClob(uint16_t index) {
	const char	*varname=obv[index].variable;
	uint32_t	length=cur->getOutputBindLength(varname);
	conn->startSendingLong(length);
	conn->sendLongSegment(cur->getOutputBindClob(varname),length);
	conn->endSendingLong();
}

bool routercursor::executeQuery(const char *query, uint32_t length,
							bool execute) {

	if (!execute) {
		return true;
	}

	if (beginquery) {
		if (routerconn->anymustbegin) {
			return begin(query,length);
		} else {
			nextrow=0;
			return true;
		}
	}

	if (!cur) {
		if (!prepareQuery(query,length)) {
			return false;
		}
	}

	if (!cur || !cur->executeQuery()) {
		return false;
	}

	checkForTempTable(query,length);

	nextrow=0;

	// populate output bind values
	for (uint16_t index=0; index<obcount; index++) {
		const char	*variable=obv[index].variable;
		*(obv[index].isnull)=routerconn->nonnullbindvalue;
		if (obv[index].type==STRING_BIND) {
			const char	*str=cur->getOutputBindString(variable);
			uint32_t	len=cur->getOutputBindLength(variable);
			if (str) {
				charstring::copy(obv[index].value.stringvalue,
								str,len);
			} else {
				obv[index].value.stringvalue[0]='\0';
				*(obv[index].isnull)=routerconn->nullbindvalue;
			} 
		} else if (obv[index].type==INTEGER_BIND) {
			*(obv[index].value.intvalue)=
					cur->getOutputBindInteger(variable);
		} else if (obv[index].type==DOUBLE_BIND) {
			*(obv[index].value.doublevalue)=
					cur->getOutputBindDouble(variable);
		}
	}

	// handle cursor bind values
	for (uint16_t index=0; index<cbcount; index++) {
		routercursor	*rcur=(routercursor *)cbv[index].cursor;
		rcur->con=con;
		rcur->cur=cur->getOutputBindCursor(cbv[index].variable);
		if (!rcur->cur) {
			return false;
		}
		rcur->cur->setResultSetBufferSize(FETCH_AT_ONCE);
		rcur->isbindcur=true;
		rcur->nextrow=0;
		if (!rcur->cur->fetchFromBindCursor()) {
			return false;
		}
	}
	return true;
}


void routercursor::checkForTempTable(const char *query, uint32_t length) {

	// for non-oracle db's
	if (charstring::compare(con->identify(),"oracle8")) {
		sqlrcursor_svr::checkForTempTable(query,length);
		return;
	}

	// for oracle db's...

	char	*ptr=(char *)query;
	char	*endptr=(char *)query+length;

	// skip any leading comments
	if (!skipWhitespace(&ptr,endptr) || !skipComment(&ptr,endptr) ||
		!skipWhitespace(&ptr,endptr)) {
		return;
	}

	// look for "create global temporary table "
	if (createoratemp.match(ptr)) {
		ptr=createoratemp.getSubstringEnd(0);
	} else {
		return;
	}

	// get the table name
	stringbuffer	tablename;
	while (ptr && *ptr && *ptr!=' ' &&
		*ptr!='\n' && *ptr!='	' && ptr<endptr) {
		tablename.append(*ptr);
		ptr++;
	}

	// append to list of temp tables
	// check for "on commit preserve rows" otherwise assume
	// "on commit delete rows"
	if (preserverows.match(ptr)) {
		conn->addSessionTempTableForTrunc(tablename.getString());
	}
}

bool routercursor::begin(const char *query, uint32_t length) {

	bool	result=true;
	for (uint16_t index=0; index<routerconn->concount; index++) {

		if (!routerconn->cons[index]) {
			continue;
		}

		// for databases that allow begin's, run the begin query,
		// for others, just set autocommit off
		bool	res=false;
		if (routerconn->beginquery[index]) {
			res=curs[index]->sendQuery(
						routerconn->beginquery[index],
						length);
			if (!res) {
				routerconn->beginQueryFailed(index);
			}
		} else {
			res=routerconn->cons[index]->autoCommitOff();
			if (!res) {
				routerconn->autoCommitOffFailed(index);
			}
		}
		if (result) {
			result=res;
			// if we had an error, set "cur" so
			// we can get the error from it
			if (!res && !cur) {
				cur=curs[index];
				curindex=0;
			}
		}
	}
	// If we're here with no "cur", then all the begin's must have
	// succeeded.  But we need a "cur" so everything else will work.
	// Any of them will do, so use the first one.
	/*if (!cur) {
		cur=curs[0];
		curindex=0;
	}*/
	return result;
}

const char *routercursor::errorMessage(bool *liveconnection) {
	// FIXME: detect downed database or downed relay
	*liveconnection=true;
	return (cur)?cur->errorMessage():"";
}

bool routercursor::knowsRowCount() {
	return true;
}

uint64_t routercursor::rowCount() {
	return (cur)?cur->rowCount():0;
}

bool routercursor::knowsAffectedRows() {
	return true;
}

uint64_t routercursor::affectedRows() {
	return (cur)?cur->affectedRows():0;
}

uint32_t routercursor::colCount() {
	return (cur)?cur->colCount():0;
}

const char * const * routercursor::columnNames() {
	return (cur)?cur->getColumnNames():0;
}

uint16_t routercursor::columnTypeFormat() {
	return (uint16_t)COLUMN_TYPE_NAMES;
}

void routercursor::returnColumnInfo() {
	if (!cur) {
		return;
	}
	for (uint32_t index=0; index<cur->colCount(); index++) {
		const char	*name=cur->getColumnName(index);
		const char	*typestring=cur->getColumnType(index);
		conn->sendColumnDefinitionString(name,
					charstring::length(name),
					typestring,
					charstring::length(typestring),
					cur->getColumnLength(index),
					cur->getColumnPrecision(index),
					cur->getColumnScale(index),
					cur->getColumnIsNullable(index),
					cur->getColumnIsPrimaryKey(index),
					cur->getColumnIsUnique(index),
					cur->getColumnIsPartOfKey(index),
					cur->getColumnIsUnsigned(index),
					cur->getColumnIsZeroFilled(index),
					cur->getColumnIsBinary(index),
					cur->getColumnIsAutoIncrement(index));
	}
}

bool routercursor::noRowsToReturn() {
	return (((cur)?cur->rowCount():0)==0);
}

bool routercursor::skipRow() {
	return fetchRow();
}

bool routercursor::fetchRow() {
	if (!cur) {
		return false;
	}
	if (cur->getField(nextrow,(uint32_t)0)) {
		nextrow++;
		return true;
	}
	return false;
}

void routercursor::returnRow() {
	if (!cur) {
		return;
	}
	for (uint32_t index=0; index<cur->colCount(); index++) {
		const char	*fld=cur->getField(nextrow-1,index);
		uint32_t	len=cur->getFieldLength(nextrow-1,index);
		if (len) {
			conn->sendField(fld,len);
		} else {
			conn->sendNullField();
		}
	}
}

void routercursor::cleanUpData(bool freeresult, bool freebinds) {
	if (freebinds) {
		if (cur) {
			cur->clearBinds();
		}
		obcount=0;
		cbcount=0;
	}
}

// FIXME: "do something" when these failures occur
void routerconnection::autoCommitOnFailed(uint16_t index) {
}

void routerconnection::autoCommitOffFailed(uint16_t index) {
}

void routerconnection::commitFailed(uint16_t index) {
}

void routerconnection::rollbackFailed(uint16_t index) {
}

void routerconnection::beginQueryFailed(uint16_t index) {
}