Asynchronous Query Execution

// AsyncRecordset.cpp : implementation file
//
#include "stdafx.h"
#include "afxconv.h"
#include "AsyncRecordset.h"

#ifdef _DEBUG
#define new DEBUG_NEW
#undef THIS_FILE
static char THIS_FILE[] = __FILE__;
#endif

/////////////////////////////////////////////////////////////////////////////
// CAsyncRecordset
static const TCHAR szDriverNotCapable[] = _T("State:S1C00");

IMPLEMENT_DYNAMIC(CAsyncRecordset, CDynamicRecordset)

CAsyncRecordset::CAsyncRecordset(CDatabase* pdb) : CDynamicRecordset(pdb)
{
	//{{AFX_FIELD_INIT(CAsyncRecordset)
	//}}AFX_FIELD_INIT
	m_nDefaultType = snapshot;
	m_bUnbound = FALSE;
}

/////////////////////////////////////////////////////////////////////////////
// CAsyncRecordset diagnostics
#ifdef _DEBUG
void CAsyncRecordset::AssertValid() const
{
	CDynamicRecordset::AssertValid();
}

void CAsyncRecordset::Dump(CDumpContext& dc) const
{
	CDynamicRecordset::Dump(dc);
}
#endif //_DEBUG

void CAsyncRecordset::OnSetOptions(HSTMT hstmt)
{
	CDynamicRecordset::OnSetOptions(hstmt);

	RETCODE nRetCode = ::SQLSetStmtOption(hstmt, SQL_ASYNC_ENABLE, SQL_ASYNC_ENABLE_ON);
	if(!Check(nRetCode))
	{
		TRACE("Warninig! Asynchronous execution not supported?");
		//ThrowDBException(nRetCode);
	}
}

// OpenAsync
BOOL CAsyncRecordset::OpenAsync(UINT nOpenType, LPCTSTR lpszSQL, DWORD dwOptions)
{
	ASSERT(!IsOpen());
	ASSERT_VALID(this);
	ASSERT(lpszSQL == NULL || ::AfxIsValidString(lpszSQL));
	ASSERT(nOpenType == AFX_DB_USE_DEFAULT_TYPE || nOpenType == dynaset || nOpenType == snapshot ||	nOpenType == forwardOnly || nOpenType == dynamic);
	ASSERT(!(dwOptions & readOnly && dwOptions & appendOnly));

	// Can only use optimizeBulkAdd with appendOnly recordsets
	ASSERT((dwOptions & optimizeBulkAdd && dwOptions & appendOnly) || !(dwOptions & optimizeBulkAdd));

	// forwardOnly recordsets have limited functionality
	ASSERT(!(nOpenType == forwardOnly && dwOptions & skipDeletedRecords));

	// Cache state info and allocate hstmt
	SetState(nOpenType, lpszSQL, dwOptions);
	if(!AllocHstmt())
		return FALSE;

	// Check if bookmarks upported (CanBookmark depends on open DB)
	ASSERT(dwOptions & useBookmarks ? CanBookmark() : TRUE);

	TRY
	{
		OnSetOptions(m_hstmt);

		// Allocate the field/param status arrays, if necessary
		m_bUnbound = FALSE;
		if(m_nFields > 0 || m_nParams > 0)
			AllocStatusArrays();
		else
			m_bUnbound = TRUE;

		// Build SQL and execute direct
		BuildSQL(lpszSQL);
		PrepareAndExecute();
	}
	CATCH_ALL(e)
	{
		Close();
		THROW_LAST();
	}
	END_CATCH_ALL

	return TRUE;
}

BOOL CAsyncRecordset::StillExecuting()
{
	USES_CONVERSION;

	ASSERT(!IsOpen());
	ASSERT_VALID(this);
	RETCODE nRetCode;
	LPCSTR lpszWSQL = T2CA(m_strSQL);

	TRY
	{
		nRetCode = ::SQLExecDirect(m_hstmt, (UCHAR*)lpszWSQL, SQL_NTS);

		// check retcode
		switch (nRetCode)
		{
			case SQL_SUCCESS_WITH_INFO:
			{
				// Check if concurrency was changed in order to mark
				// recordset non-updatable if necessary
				DWORD dwConcurrency;
				AFX_SQL_SYNC(::SQLGetStmtOption(m_hstmt, SQL_CONCURRENCY, &dwConcurrency));
				if(!Check(nRetCode))
					ThrowDBException(nRetCode);

				if (dwConcurrency == SQL_CONCUR_READ_ONLY && (m_bUpdatable || m_bAppendable))
				{
					m_bUpdatable = FALSE;
					m_bAppendable = FALSE;

	#ifdef _DEBUG
					if (afxTraceFlags & traceDatabase)
					{
						TRACE0("Warning: Concurrency changed by driver.\n");
						TRACE0("\tMarking CRecordset as not updatable.\n");
					}
	#endif // _DEBUG
				}

				break;
			}
			case SQL_SUCCESS:
			case SQL_NO_DATA_FOUND:
				break;
			case SQL_STILL_EXECUTING:
				return TRUE;
				break;
			default:
				ThrowDBException(nRetCode);
		}

		m_lOpen = AFX_RECORDSET_STATUS_OPEN;

		// Cache some field info and prepare the rowset
		AllocAndCacheFieldInfo();
		AllocRowset();

		// If late binding, still need to allocate status arrays
		if(m_bUnbound && (m_nFields > 0 || m_nParams > 0))
			AllocStatusArrays();

		// Give derived classes a call before binding
		PreBindFields();

		// Fetch the first row of data
		MoveNext();

		// If EOF, then result set empty, so set BOF as well
		m_bBOF = m_bEOF;
	}
	CATCH_ALL(e)
	{
		Close();
		THROW_LAST();
	}
	END_CATCH_ALL
	
	return FALSE;
}

void CAsyncRecordset::PrepareAndExecute()
{
	USES_CONVERSION;
	RETCODE nRetCode;
	LPCSTR lpszWSQL = T2CA(m_strSQL);
	BOOL bConcurency = FALSE;

	while(!bConcurency)
	{
		nRetCode = ::SQLExecDirect(m_hstmt, (UCHAR*)lpszWSQL, SQL_NTS);

		// check retcode
		switch (nRetCode)
		{
			case SQL_SUCCESS_WITH_INFO:
			case SQL_SUCCESS:
			case SQL_NO_DATA_FOUND:
			case SQL_STILL_EXECUTING:
				bConcurency = TRUE;
				break;
			default:
			{
				// If "Driver Not Capable" error, assume cursor type doesn't
				// support requested concurrency and try alternate concurrency.
				CDBException* e = new CDBException(nRetCode);
				e->BuildErrorString(m_pDatabase, m_hstmt);
				if(m_dwConcurrency != SQL_CONCUR_READ_ONLY && e->m_strStateNativeOrigin.Find(szDriverNotCapable) >= 0)
				{
#ifdef _DEBUG
					if (afxTraceFlags & traceDatabase)
						TRACE0("Warning: Driver does not support requested concurrency.\n");
#endif
					// Don't need exception to persist while attempting to reset concurrency
					e->Delete();

					// ODBC will automatically attempt to set alternate concurrency if
					// request fails, but it won't try LOCK even if driver supports it.
					if((m_dwDriverConcurrency & SQL_SCCO_LOCK) && (m_dwConcurrency == SQL_CONCUR_ROWVER || m_dwConcurrency == SQL_CONCUR_VALUES))
					{
						m_dwConcurrency = SQL_CONCUR_LOCK;
					}
					else
					{
						m_dwConcurrency = SQL_CONCUR_READ_ONLY;
						m_bUpdatable = m_bAppendable = FALSE;
#ifdef _DEBUG
						if (afxTraceFlags & traceDatabase)
							TRACE0("Warning: Setting recordset read only.\n");
#endif
					}

					// Attempt to reset the concurrency model.
					AFX_SQL_SYNC(::SQLSetStmtOption(m_hstmt, SQL_CONCURRENCY, m_dwConcurrency));
					if(!Check(nRetCode))
					{
						TRACE0("Error: ODBC failure setting recordset concurrency.\n");
						ThrowDBException(nRetCode);
					}
				}
				else
				{
					TRACE0("Error: ODBC failure on SQLPrepare or SQLExecDirect\n");
					THROW(e);
				}
			}
		}
	}
}