myclient.cpp

通用的数据库中间库

		sServerName = sDBString.Left(iPos);
		sDatabase = sDBString.Mid(iPos+1);
	}
	else
		sDatabase = sDBString;

	iPos = sServerName.Find('/');
	if(iPos > 0) // unix_socket
		sUnixSocket = sServerName;
	else
	{
		iPos = sServerName.Find(':');
		if(iPos > 0) // alternate port number found
		{
			sHost = sServerName.Left(iPos);
			port = atoi(sServerName.Mid(iPos+1));
		}
		else
			sHost = sServerName;
	}

	m_handles.mysql = g_myAPI.mysql_init(NULL);
	if(!m_handles.mysql)
		throw SAException(SA_Library_Error, -1, -1, "mysql_init -> NULL");

	try
	{
		if(g_myAPI.mysql_real_connect2) // version with database name
		{
			if(!g_myAPI.mysql_real_connect2(
				m_handles.mysql,
				sHost.IsEmpty()? (const char *)NULL : (const char *)sHost,
				sUserID.IsEmpty()? (const char *)NULL : (const char *)sUserID,
				sPassword.IsEmpty()? (const char *)NULL : (const char *)sPassword,
				sDatabase.IsEmpty()? (const char *)NULL : (const char *)sDatabase,
				port,
				sUnixSocket.IsEmpty()? (const char*)NULL : (const char *)sUnixSocket,
				0))
				Check(m_handles.mysql);
		}
		else
		{
			if(!g_myAPI.mysql_real_connect1(
				m_handles.mysql,
				sHost.IsEmpty()? (const char *)NULL : (const char *)sHost,
				sUserID.IsEmpty()? (const char *)NULL : (const char *)sUserID,
				sPassword.IsEmpty()? (const char *)NULL : (const char *)sPassword,
				port,
				sUnixSocket.IsEmpty()? (const char *)NULL : (const char *)sUnixSocket,
				0))
				Check(m_handles.mysql);
			if(!sDatabase.IsEmpty() && g_myAPI.mysql_select_db(m_handles.mysql, sDatabase))
				Check(m_handles.mysql);
		}
	}
	catch(SAException &)	// clean up
	{
		if(m_handles.mysql)
		{
			g_myAPI.mysql_close(m_handles.mysql);
			m_handles.mysql = NULL;
		}

		throw;
	}
}

/*virtual */
void ImyConnection::Disconnect()
{
	assert(m_handles.mysql != NULL);

	g_myAPI.mysql_close(m_handles.mysql);
	m_handles.mysql = NULL;
}

/*virtual */
void ImyConnection::Commit()
{
	if(g_myAPI.mysql_query(m_handles.mysql, "COMMIT"))
		Check(m_handles.mysql);
}

/*virtual */
void ImyConnection::Rollback()
{
	if(g_myAPI.mysql_query(m_handles.mysql, "ROLLBACK"))
		Check(m_handles.mysql);
}

/*virtual */
saAPI *ImyConnection::NativeAPI() const
{
	return 	&g_myAPI;
}

/*virtual */
saConnectionHandles *ImyConnection::NativeHandles()
{
	return &m_handles;
}

/*virtual */
void ImyConnection::setIsolationLevel(
	SAIsolationLevel_t eIsolationLevel)
{
	// Setting the SESSION privilege will affect the following and all future transactions.
	switch(eIsolationLevel)
	{
	case SA_ReadUncommitted:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED"))
			Check(m_handles.mysql);
		break;
	case SA_ReadCommitted:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED"))
			Check(m_handles.mysql);
		break;
	case SA_RepeatableRead:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ"))
			Check(m_handles.mysql);
		break;
	case SA_Serializable:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE"))
			Check(m_handles.mysql);
		break;
	default:
		assert(false);
		return;
	}
	// start new transaction
	m_pSAConnection->Commit();
}

/*virtual */
void ImyConnection::setAutoCommit(
	SAAutoCommit_t eAutoCommit)
{
	switch(eAutoCommit)
	{
	case SA_AutoCommitOff:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET AUTOCOMMIT=0"))
			Check(m_handles.mysql);
		break;
	case SA_AutoCommitOn:
		if(g_myAPI.mysql_query(m_handles.mysql, "SET AUTOCOMMIT=1"))
			Check(m_handles.mysql);
		break;
	default:
		assert(false);
	}
}

/*virtual */
ISACursor *ImyConnection::NewCursor(SACommand *m_pCommand)
{
	return new ImyCursor(this, m_pCommand);
}

//////////////////////////////////////////////////////////////////////
// ImyCursor implementation
//////////////////////////////////////////////////////////////////////

/*virtual */
bool ImyCursor::IsOpened()
{
	return m_bOpened;
}

/*virtual */
void ImyCursor::Open()
{
	assert(m_bOpened == false);

	m_bOpened = true;
}

/*virtual */
void ImyCursor::Close()
{
	assert(m_bOpened == true);

	m_bOpened = false;
}

// prepare statement (also convert to native format if needed)
/*virtual */
void ImyCursor::Prepare(
	const SAString &sStmt,
	SACommandType_t/* eCmdType*/,
	int/* nPlaceHolderCount*/,
	saPlaceHolder ** /*ppPlaceHolders*/)
{
	// save original statement
	m_sOriginalStmst = sStmt;
}

/*virtual */
void ImyCursor::UnExecute()
{
	if(m_handles.result)
	{
		g_myAPI.mysql_free_result(m_handles.result);
		m_handles.result = NULL;
	}
	m_bResultSetCanBe = false;
}

// executes statement (also binds parameters if needed)
/*virtual */
void ImyCursor::Execute(
	int nPlaceHolderCount,
	saPlaceHolder **ppPlaceHolders)
{
	SAString sBoundStmt;
	// change ':' param markers to bound values
	unsigned int nPos = 0;
	for(int i = 0; i < nPlaceHolderCount; ++i)
	{
		SAParam &Param = *ppPlaceHolders[i]->getParam();

		sBoundStmt += m_sOriginalStmst.Mid(nPos, ppPlaceHolders[i]->getStart()-nPos);

		if(Param.isNull())
		{
			sBoundStmt += "NULL";
		}
		else
		{
			SAString sBoundValue;
			SAString sTemp;
			switch(Param.DataType())
			{
			case SA_dtUnknown:
				throw SAException(SA_Library_Error, -1, -1, IDS_UNKNOWN_PARAMETER_TYPE, (const char*)Param.Name());
			case SA_dtBool:
				sBoundValue = Param.asBool()? "1" : "0";
				break;
			case SA_dtShort:
				sBoundValue = Param.asString();	// use SQLAPI++ converter
				break;
			case SA_dtLong:
				sBoundValue = Param.asString();	// use SQLAPI++ converter
				break;
			case SA_dtDouble:
				sBoundValue = Param.asString();	// use SQLAPI++ converter
				break;
			case SA_dtDateTime:
				ImyConnection::CnvtDateTimeToInternal(
					Param.setAsDateTime(), sTemp);
				sBoundValue = "'";
				sBoundValue += sTemp;
				sBoundValue += "'";
				break;
			case SA_dtString:
				sBoundValue = "'";
				sBoundValue += MySQLEscapeString(Param.asString());
				sBoundValue += "'";
				break;
			case SA_dtBytes:
				sBoundValue = "'";
				sBoundValue += MySQLEscapeString(Param.asBytes());
				sBoundValue += "'";
				break;
			case SA_dtLongBinary:
			case SA_dtBLob:
				BindBLob(Param, sBoundStmt);
				break;
			case SA_dtLongChar:
			case SA_dtCLob:
				BindText(Param, sBoundStmt);
				break;
			default:
				assert(false);
			}

			switch(Param.DataType())
			{
			case SA_dtLongBinary:
			case SA_dtLongChar:
			case SA_dtBLob:
			case SA_dtCLob:
				break;	// was already written
			default:
				sBoundStmt += sBoundValue;
			}
		}
		nPos = ppPlaceHolders[i]->getEnd() + 1;
	}
	// copy tail
	if(nPos < strlen(m_sOriginalStmst))
		sBoundStmt += m_sOriginalStmst.Mid(nPos);

	myConnectionHandles *pConH = (myConnectionHandles *)m_pCommand->Connection()->NativeHandles();
	if(g_myAPI.mysql_real_query(pConH->mysql, sBoundStmt, sBoundStmt.GetLength()))
		Check(pConH->mysql);

	SAString sOption = m_pCommand->Option("HandleResult");
	if(sOption.CompareNoCase("store") == 0)
		m_handles.result = g_myAPI.mysql_store_result(pConH->mysql);
	else
		m_handles.result = g_myAPI.mysql_use_result(pConH->mysql);

	m_bResultSetCanBe = true;
		
	Check(pConH->mysql);
}

SAString ImyCursor::MySQLEscapeString(const char *sFrom, int nLen)
{
	SAString sNew;

	char *p = sNew.GetBuffer(nLen*2);
	unsigned int nNewLen;
	if(g_myAPI.mysql_real_escape_string)
	{
		myConnectionHandles *pConH = (myConnectionHandles *)m_pCommand->Connection()->NativeHandles();
		nNewLen = g_myAPI.mysql_real_escape_string(
			pConH->mysql,
			p,
			sFrom,
			(unsigned int)nLen);
	}
	else
		nNewLen = g_myAPI.mysql_escape_string(
			p,
			sFrom,
			(unsigned int)nLen);
	sNew.ReleaseBuffer((int)nNewLen);

	return sNew;
}

SAString ImyCursor::MySQLEscapeString(const SAString &sValue)
{
	int nLen = sValue.GetLength();
	if(!nLen)
		return sValue;

	return MySQLEscapeString(sValue, nLen);
}

void ImyCursor::BindBLob(SAParam &Param, SAString &sBoundStmt)
{
	sBoundStmt += "'";

	unsigned int nActualWrite;
	SAPieceType_t ePieceType = SA_FirstPiece;
	void *pBuf;
	while((nActualWrite = Param.InvokeWriter(
		ePieceType, ImyConnection::MaxLongPiece, pBuf)) != 0)
	{
		sBoundStmt += MySQLEscapeString((const char *)pBuf, nActualWrite);

		if(ePieceType == SA_LastPiece)
			break;
	}

	sBoundStmt += "'";
}

void ImyCursor::BindText(SAParam &Param, SAString &sBoundStmt)
{
	sBoundStmt += "'";

	unsigned int nActualWrite;
	SAPieceType_t ePieceType = SA_FirstPiece;
	void *pBuf;
	while((nActualWrite = Param.InvokeWriter(
		ePieceType, ImyConnection::MaxLongPiece, pBuf)) != 0)
	{
		sBoundStmt += MySQLEscapeString((const char *)pBuf, nActualWrite);

		if(ePieceType == SA_LastPiece)
			break;
	}

	sBoundStmt += "'";
}

/*virtual */
void ImyCursor::Cancel()
{
	myConnectionHandles *pConH = (myConnectionHandles *)m_pCommand->Connection()->NativeHandles();
	// kill connection
	if(g_myAPI.mysql_kill(pConH->mysql, g_myAPI.mysql_thread_id(pConH->mysql)))
		Check(pConH->mysql);

	// try to restore connection
	g_myAPI.mysql_ping(pConH->mysql);
}

/*virtual */
bool ImyCursor::ResultSetExists()
{
	if(!m_bResultSetCanBe)
		return false;

	return m_handles.result != NULL;
}

/*virtual */
void ImyCursor::DescribeFields(
	DescribeFields_cb_t fn)
{
	myConnectionHandles *pConH = (myConnectionHandles *)m_pCommand->Connection()->NativeHandles();

	unsigned int cFields = 0;
	if(g_myAPI.mysql_field_count)
		cFields = g_myAPI.mysql_field_count(pConH->mysql);
	else
		assert(false);

	MYSQL_FIELD *fields;
	fields = g_myAPI.mysql_fetch_fields(m_handles.result);
	for(unsigned int iField = 0; iField < cFields; ++iField)
	{
		(m_pCommand->*fn)(
			fields->name,
			ImyConnection::CnvtNativeToStd(
				fields->type,
				fields->length,
				fields->decimals,
				fields->flags),
			fields->type,
			fields->length,
			0,
			0,
			false);

		++fields;
	}
}

/*virtual */
void ImyCursor::SetSelectBuffers()
{
	// do nothing
}

/*virtual */
void ImyCursor::SetFieldBuffer(
	int/* nCol*/,	// 1-based
	void * /*pInd*/,
	unsigned int/* nIndSize*/,
	void * /*pSize*/,
	unsigned int/* nSizeSize*/,
	void * /*pValue*/,
	unsigned int/* nValueSize*/)
{
	assert(false);	// we do not use standard allocators
}

void ImyCursor::ConvertMySQLRowToFields()
{
	int cFields = m_pCommand->FieldCount();
	for(int iField = 0; iField < cFields; ++iField)
	{
		SAField &Field = m_pCommand->Field(iField+1);
		SADataType_t eFieldType = Field.FieldType();

		if(m_mysql_row[iField] == NULL)
		{
			*Field.m_pbNull = true;
			continue;
		}
		*Field.m_pbNull = false;

		const char *sValue = m_mysql_row[iField];
		unsigned int nRealSize = m_lengths[iField];

		switch(eFieldType)
		{
		case SA_dtUnknown:
			throw SAException(SA_Library_Error, -1, -1, IDS_UNKNOWN_DATA_TYPE);
		case SA_dtShort:
			Field.m_eDataType = SA_dtShort;
			*(short*)Field.m_pScalar = (short)atol(sValue);
			break;
		case SA_dtLong:
			Field.m_eDataType = SA_dtLong;
			*(long*)Field.m_pScalar = atol(sValue);
			break;
		case SA_dtDouble:
			Field.m_eDataType = SA_dtDouble;
			*(double*)Field.m_pScalar = atof(sValue);
			break;
		case SA_dtDateTime:
			Field.m_eDataType = SA_dtDateTime;
			ImyConnection::CnvtInternalToDateTime(
				*Field.m_pDateTime,
				sValue);
			break;
		case SA_dtString:
			Field.m_eDataType = SA_dtString;
			*Field.m_pString =
				SAString(sValue, nRealSize);
			break;
		case SA_dtBytes:
			Field.m_eDataType = SA_dtBytes;
			*Field.m_pString =
				SAString(sValue, nRealSize);
			break;
		case SA_dtLongBinary:
			Field.m_eDataType = SA_dtLongBinary;
			break;
		case SA_dtLongChar:
			Field.m_eDataType = SA_dtLongChar;
			break;
		case SA_dtBLob:
			Field.m_eDataType = SA_dtBLob;
			break;
		case SA_dtCLob:
			Field.m_eDataType = SA_dtCLob;
			break;
		default:
			assert(false);	// unknown type
		}

		if(isLongOrLob(eFieldType))
			ConvertLongOrLOB(ISA_FieldValue, Field, NULL, 0);
	}
}

/*virtual */
bool ImyCursor::FetchNext()
{
	m_mysql_row = g_myAPI.mysql_fetch_row(m_handles.result);

	if(m_mysql_row)
	{
		m_lengths = g_myAPI.mysql_fetch_lengths(m_handles.result);
		ConvertMySQLRowToFields();
	}
	else
		m_bResultSetCanBe = false;

	return m_mysql_row != NULL;
}

/*virtual */
long ImyCursor::GetRowsAffected()
{
	myConnectionHandles *pConH = (myConnectionHandles *)m_pCommand->Connection()->NativeHandles();

	my_ulonglong cRows = g_myAPI.mysql_affected_rows(pConH->mysql);
	
	return (long)cRows;
}

/*virtual */
void ImyCursor::ReadLongOrLOB(
	ValueType_t/* eValueType*/,
	SAValueRead &vr,
	void * /*pValue*/,
	unsigned int/* nFieldBufSize*/,
	saLongOrLobReader_t fnReader,
	unsigned int nReaderWantedPieceSize,
	void *pAddlData)
{
	int nPos = ((SAField &)vr).Pos();

	// get long data and size
	const char *pData = m_mysql_row[nPos-1];
	unsigned long nLongSize = m_lengths[nPos-1];

	unsigned char* pBuf;
	unsigned int nPieceSize = vr.PrepareReader(
		nLongSize,
		ImyConnection::MaxLongPiece,
		pBuf,
		fnReader,
		nReaderWantedPieceSize,
		pAddlData);
	assert(nPieceSize <= ImyConnection::MaxLongPiece);

	SAPieceType_t ePieceType = SA_FirstPiece;
	unsigned long nTotalRead = 0;
	do
	{
		nPieceSize =
			sa_min(nPieceSize, nLongSize - nTotalRead);

		memcpy(pBuf, pData + nTotalRead, nPieceSize);
		unsigned int actual_read = nPieceSize;

		nTotalRead += actual_read;
		assert(nTotalRead <= nLongSize);

        if(nTotalRead == nLongSize)
        {
            if(ePieceType == SA_NextPiece)
                ePieceType = SA_LastPiece;
            else    // the whole Long was read in one piece
            {
                assert(ePieceType == SA_FirstPiece);
                ePieceType = SA_OnePiece;
            }
        }
		vr.InvokeReader(ePieceType, pBuf, actual_read);

		if(ePieceType == SA_FirstPiece)
			ePieceType = SA_NextPiece;
	}
	while(nTotalRead < nLongSize);
}

/*virtual */
void ImyCursor::DescribeParamSP()
{
	assert(false);	// no SPs in MySQL
}

/*virtual */
saCommandHandles *ImyCursor::NativeHandles()
{
	return &m_handles;
}

//////////////////////////////////////////////////////////////////////
// MySQL globals
//////////////////////////////////////////////////////////////////////

ISAConnection *newImyConnection(SAConnection *pSAConnection)
{
	return new ImyConnection(pSAConnection);
}