pgclient.cpp

通用的数据库中间库

/*static */
void IpgConnection::CnvtInternalToDateTime(
	SADateTime &date_time,
	const char *sInternal)
{
	// first initialize to default date/time values
	SADateTime dt;
	int nYear = dt.GetYear();
	int nMonth = 1;
	int nDay = 1;
	int nHour = 0;
	int nMinute = 0;
	int nSecond = 0;
	int nNanoSecond = 0;
	int nZoneHour = 0;

	// Available date time formats
	// ISO		1997-12-17 07:37:16-08 
	// SQL		12/17/1997 07:37:16.00 PST 
	// Postgres	Wed Dec 17 07:37:16 1997 PST 
	// German	17.12.1997 07:37:16.00 PST 

	if( ::strlen(sInternal) < 19 ) // date only or time only
	{
		if( strchr(sInternal, ':') ) // time
			ParseInternalTime(sInternal, nHour, nMinute, nSecond, nNanoSecond, nZoneHour);
		else // date
			ParseInternalDate(sInternal, nYear, nMonth, nDay);
	}
	else // datetime
	{
		if( isdigit(sInternal[0]) ) // ISO, SQL or German
		{
			ParseInternalDate(sInternal, nYear, nMonth, nDay);
			assert(strchr( sInternal, ' '));
			ParseInternalTime(sInternal+11, nHour, nMinute, nSecond, nNanoSecond, nZoneHour);
		}
		else // Postgres format
		{
			// TODO:
			assert(false);
		}
	}

	if( nZoneHour != 0 )
	{
	}

	if(nMonth != 0 && nDay != 0 && nHour <=23)	// simple test for validness
		date_time = SADateTime(nYear, nMonth, nDay, nHour, nMinute, nSecond);
	else
		date_time = dt;

	date_time.Fraction() = nNanoSecond;
}

/*static */
void IpgConnection::CnvtDateTimeToInternal(
	const SADateTime &date_time,
	SAString &sInternal)
{
	// Datetime's format is YYYY-MM-DD HH:MM:SS.fraction
	sInternal.Format("%.4d-%.2d-%.2d %.2d:%.2d:%.2d.%.3ld",
		date_time.GetYear(), date_time.GetMonth(), date_time.GetDay(),
		date_time.GetHour(), date_time.GetMinute(), date_time.GetSecond(),
		date_time.Fraction()/1000000);
}

/*virtual */
void IpgConnection::CnvtInternalToCursor(
	SACommand * /*pCursor*/,
	const void * /*pInternal*/)
{
	assert(false);
}

/*virtual */
long IpgConnection::GetClientVersion() const
{
	return g_nPostgreSQLDLLVersionLoaded;
}

/*virtual */
long IpgConnection::GetServerVersion() const
{
	return SAExtractVersionFromString(GetServerVersionString());
}

/*virtual */
SAString IpgConnection::GetServerVersionString() const
{
	SAString sVersion;

	if( IsConnected() )
	{
		PGresult *res = g_pgAPI.PQexec(m_handles.conn, "select version()");
		Check(res);
		sVersion = g_pgAPI.PQgetvalue(res, 0, 0);
	}

	return  sVersion;
}

/*virtual */
bool IpgConnection::IsConnected() const
{
	return m_handles.conn != NULL;
}

/*virtual */
void IpgConnection::Connect(
	const SAString &sDBString,
	const SAString &sUserID,
	const SAString &sPassword)
{
	assert(m_handles.conn == NULL);

	// dbstring as: [server_name][@][dbname]
	// server_name as: hostname[:port], or unix_socket path
	// for connection to server without dbname use 'server_name@'

	SAString sServerName, sDatabase, sHost, sPort;

	int iPos = sDBString.Find('@');
	if(iPos >= 0) // Database is present in connection string
	{
		sServerName = sDBString.Left(iPos);
		sDatabase = sDBString.Mid(iPos+1);
	}
	else
		sDatabase = sDBString;

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

	m_handles.conn = g_pgAPI.PQsetdbLogin(
		sHost.IsEmpty()? (const char *)NULL : (const char *)sHost,
		sPort.IsEmpty()? (const char *)NULL : (const char *)sPort,
        NULL, NULL,
		sDatabase.IsEmpty()? (const char *)NULL : (const char *)sDatabase,
		sUserID.IsEmpty()? (const char *)NULL : (const char *)sUserID,
		sPassword.IsEmpty()? (const char *)NULL : (const char *)sPassword);

	if( g_pgAPI.PQstatus(m_handles.conn) == CONNECTION_BAD )
		throw SAException(SA_RDBMS_API_Error, CONNECTION_BAD, -1, g_pgAPI.PQerrorMessage(m_handles.conn));
}

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

	g_pgAPI.PQfinish(m_handles.conn);
	m_handles.conn = NULL;
}

/*virtual */
void IpgConnection::Commit()
{
	Check(g_pgAPI.PQexec(m_handles.conn, "COMMIT"));
	// and start new transaction if not in autocommit mode
	if( m_pSAConnection->AutoCommit() == SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(m_handles.conn, "BEGIN"));
}

/*virtual */
void IpgConnection::Rollback()
{
	Check(g_pgAPI.PQexec(m_handles.conn, "ROLLBACK"));
	// and start new transaction if not in autocommit mode
	if( m_pSAConnection->AutoCommit() == SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(m_handles.conn, "BEGIN"));
}

/*virtual */
saAPI *IpgConnection::NativeAPI() const
{
	return 	&g_pgAPI;
}

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

/*virtual */
void IpgConnection::setIsolationLevel(
	SAIsolationLevel_t eIsolationLevel)
{
	SAString sCmd("SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL ");
	
	if( eIsolationLevel == SA_Serializable )
		sCmd += "SERIALIZABLE";
	else
		sCmd += "READ COMMITTED";

	Check(g_pgAPI.PQexec(m_handles.conn, sCmd));
}

/*virtual */
void IpgConnection::setAutoCommit(
	SAAutoCommit_t eAutoCommit)
{
	Check(g_pgAPI.PQexec(m_handles.conn, "COMMIT"));
	if( eAutoCommit == SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(m_handles.conn, "BEGIN"));
}

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

//////////////////////////////////////////////////////////////////////
// IpgCursor implementation
//////////////////////////////////////////////////////////////////////

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

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

	m_bOpened = true;
}

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

	m_bOpened = false;
}

// prepare statement (also convert to native format if needed)
/*virtual */
void IpgCursor::Prepare(
	const SAString &/*sStmt*/,
	SACommandType_t/* eCmdType*/,
	int/* nPlaceHolderCount*/,
	saPlaceHolder ** /*ppPlaceHolders*/)
{
	// no prepare in libpq
	// all is beeing done in Execute()
}

/*virtual */
void IpgCursor::UnExecute()
{
	if(m_handles.res)
	{
		g_pgAPI.PQclear(m_handles.res);
		m_handles.res = NULL;
	}

	m_bResultSetCanBe = false;
}

// executes statement (also binds parameters if needed)
/*virtual */
void IpgCursor::Execute(
	int nPlaceHolderCount,
	saPlaceHolder **ppPlaceHolders)
{
	m_nCurrentTuple = 0;
	m_nTuplesCount = 0;

	SAString sOriginalStmst = m_pCommand->CommandText();
	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 += 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() ? "TRUE" : "FALSE";
				break;
			case SA_dtShort:
			case SA_dtLong:
			case SA_dtDouble:
				sBoundValue = Param.asString();	// use SQLAPI++ converter
				break;
			case SA_dtDateTime:
				IpgConnection::CnvtDateTimeToInternal(
					Param.setAsDateTime(), sTemp);
				sBoundValue = "'";
				sBoundValue += sTemp;
				sBoundValue += "'";
				break;
			case SA_dtBytes:
				{
					sBoundValue = "'";
					SAString sValue = Param.asBytes();
					char* szByteStr = IpgConnection::byte2string((const void*)sValue, sValue.GetBinaryLength() );
					sBoundValue += szByteStr;
					free(szByteStr);
					sBoundValue += "'";
				}
				break;
			case SA_dtString: // Quote ' with '
				sTemp = Param.asString();
				sBoundValue = "'";
				sBoundValue += IpgConnection::EscapeString(sTemp);
				sBoundValue += "'";
				break;
			case SA_dtLongChar:
				BindLongChar(Param, sBoundStmt);
				break;
			case SA_dtLongBinary:
				BindLongBinary(Param, sBoundStmt);
				break;
			case SA_dtBLob:
			case SA_dtCLob:
				BindBLob(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(sOriginalStmst))
		sBoundStmt += sOriginalStmst.Mid(nPos);

	pgConnectionHandles *pConH = (pgConnectionHandles *)m_pCommand->Connection()->NativeHandles();

	try
	{
		Check(m_handles.res = g_pgAPI.PQexec(pConH->conn, sBoundStmt));
	}
	catch( SAException& )  // Cleanup
	{
		if(m_handles.res)
		{
			g_pgAPI.PQclear(m_handles.res);
			m_handles.res = NULL;
		}

		throw;
	}


	if( PGRES_TUPLES_OK  == g_pgAPI.PQresultStatus(m_handles.res) )
	{
		m_nTuplesCount = g_pgAPI.PQntuples(m_handles.res);
		m_bResultSetCanBe = true;
	}
	else
		m_bResultSetCanBe = false;
}

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

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

		if(ePieceType == SA_LastPiece)
			break;
	}

	sBoundStmt += "'";
}

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

	unsigned int nActualWrite;
	SAPieceType_t ePieceType = SA_FirstPiece;
	void *pBuf;
	while((nActualWrite = Param.InvokeWriter(
		ePieceType, IpgConnection::MaxLongPiece, pBuf)) != 0)
	{
		char* szByteStr = IpgConnection::byte2string(pBuf, nActualWrite);
		sBoundStmt += szByteStr;
		free(szByteStr);

		if(ePieceType == SA_LastPiece)
			break;
	}

	sBoundStmt += "'";
}

void IpgCursor::BindBLob(SAParam &Param, SAString &sBoundStmt)
{
	unsigned int nActualWrite;
	SAPieceType_t ePieceType = SA_FirstPiece;
	void *pBuf;
	pgConnectionHandles *pConH = (pgConnectionHandles *)m_pCommand->Connection()->NativeHandles();
	SAConnection *pSAConnection = m_pCommand->Connection();

	// Start transaction if not started
	if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(pConH->conn, "BEGIN"));

	Oid blobOid = g_pgAPI.lo_creat(pConH->conn, INV_READ|INV_WRITE);
	int fd = g_pgAPI.lo_open(pConH->conn, blobOid, INV_WRITE);

	while((nActualWrite = Param.InvokeWriter(
		ePieceType, IpgConnection::MaxBlobPiece, pBuf)) != 0)
	{
		char *pWriteBuf = (char*)pBuf;
		int nToWrite = nActualWrite;

		do
		{
			int nWritten;

			if( (nWritten = g_pgAPI.lo_write(pConH->conn, fd, pWriteBuf, nToWrite)) < 0 )
			{
				g_pgAPI.lo_close(pConH->conn, fd);
				
				// Rollback transaction, if previos start it
				if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
					g_pgAPI.PQexec(pConH->conn, "ROLLBACK");
				
				throw SAException(SA_Library_Error, -1, -1, "lo_write -> negative number");
			}

			nToWrite -= nWritten;
			pWriteBuf += nWritten;
		}
		while( nToWrite > 0 );

		if(ePieceType == SA_LastPiece)
			break;
	}

	g_pgAPI.lo_close(pConH->conn, fd);

	// End transaction, if previos start it
	if( pSAConnection->AutoCommit() != SA_AutoCommitOff )
		Check(g_pgAPI.PQexec(pConH->conn, "END"));

	SAString sBlobOid; sBlobOid.Format("%d", blobOid);
	sBoundStmt += sBlobOid;
}

/*static*/
SAString IpgConnection::EscapeString(const char* szString)
{
	SAString sReturn(szString);
	sReturn.Replace("\\", "\\\\");
	sReturn.Replace("'", "\\'");
	return sReturn;
}

#define VAL(CH)			((CH) - '0')
#define DIG(VAL)		((VAL) + '0')

/*static*/
char* IpgConnection::byte2string(const void* pByte, int nBufLen)
{
	// Non-printable and '\' characters are inserted as '\nnn' (octal)
	// and ''' as '\''.

	int i, len = 1;
	char *vp = (char*)pByte;

	for( i = nBufLen; i != 0; i--, vp++)
	{
		if( *vp == '\'' )
			len += 2;
		else if( *vp == 0 )
			len += 5;
		else if( isprint((unsigned char) *vp) && *vp != '\\')
			len++;
		else
			len += 4;
	}

	char *result = (char*)malloc(len);

	char *rp = result;
	vp = (char*)pByte;
	
	int val;			/* holds unprintable chars */
	for( i = nBufLen; i != 0; i--, vp++ )