src/qpid/store/ms-sql/MessageMapRecordset.cpp - qpid-cpp - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */

 #include <qpid/Exception.h>
 #include <qpid/log/Statement.h>
 #include <qpid/store/StorageProvider.h>

 #include "MessageMapRecordset.h"
 #include "BlobEncoder.h"
 #include "DatabaseConnection.h"
 #include "Exception.h"
 #include "VariantHelper.h"

 namespace {
 inline void TESTHR(HRESULT x) {if FAILED(x) _com_issue_error(x);};
 }

 namespace qpid {
 namespace store {
 namespace ms_sql {

 void
 MessageMapRecordset::open(DatabaseConnection* conn, const std::string& table)
 {
     init(conn, table);
 }

 void
 MessageMapRecordset::add(uint64_t messageId,
                          uint64_t queueId,
                          const std::string& xid)
 {
     std::ostringstream command;
     command << "INSERT INTO " << tableName
             << " (messageId, queueId";
     if (!xid.empty())
         command << ", prepareStatus, xid";
     command << ") VALUES (" << messageId << "," << queueId;
     if (!xid.empty())
         command << "," << PREPARE_ADD << ",?";
     command << ")" << std::ends;

     _CommandPtr cmd = NULL;
     _ParameterPtr xidVal = NULL;
     TESTHR(cmd.CreateInstance(__uuidof(Command)));
     _ConnectionPtr p = *dbConn;
     cmd->ActiveConnection = p;
     cmd->CommandText = command.str().c_str();
     cmd->CommandType = adCmdText;
     if (!xid.empty()) {
         TESTHR(xidVal.CreateInstance(__uuidof(Parameter)));
         xidVal->Name = "@xid";
         xidVal->Type = adVarBinary;
         xidVal->Size = xid.length();
         xidVal->Direction = adParamInput;
         xidVal->Value = BlobEncoder(xid);
         cmd->Parameters->Append(xidVal);
     }
     cmd->Execute(NULL, NULL, adCmdText | adExecuteNoRecords);
 }

 void
 MessageMapRecordset::remove(uint64_t messageId, uint64_t queueId)
 {
     std::ostringstream command;
     command << "DELETE FROM " << tableName
             << " WHERE queueId = " << queueId
             << " AND messageId = " << messageId << std::ends;
     _CommandPtr cmd = NULL;
     TESTHR(cmd.CreateInstance(__uuidof(Command)));
     _ConnectionPtr p = *dbConn;
     cmd->ActiveConnection = p;
     cmd->CommandText = command.str().c_str();
     cmd->CommandType = adCmdText;
     _variant_t deletedRecords;
     cmd->Execute(&deletedRecords, NULL, adCmdText | adExecuteNoRecords);
     if ((long)deletedRecords == 0)
         throw ms_sql::Exception("Message does not exist in queue mapping");
     // Trigger on deleting the mapping takes care of deleting orphaned
     // message record from tblMessage.
 }

 void
 MessageMapRecordset::pendingRemove(uint64_t messageId,
                                    uint64_t queueId,
                                    const std::string& xid)
 {
     // Look up the mapping for the specified message and queue. There
     // should be only one because of the uniqueness constraint in the
     // SQL table. Update it to reflect it's pending delete with
     // the specified xid.
     std::ostringstream command;
     command << "UPDATE " << tableName
             << " SET prepareStatus=" << PREPARE_REMOVE
             << " , xid=?"
             << " WHERE queueId = " << queueId
             << " AND messageId = " << messageId << std::ends;

     _CommandPtr cmd = NULL;
     _ParameterPtr xidVal = NULL;
     TESTHR(cmd.CreateInstance(__uuidof(Command)));
     TESTHR(xidVal.CreateInstance(__uuidof(Parameter)));
     _ConnectionPtr p = *dbConn;
     cmd->ActiveConnection = p;
     cmd->CommandText = command.str().c_str();
     cmd->CommandType = adCmdText;
     xidVal->Name = "@xid";
     xidVal->Type = adVarBinary;
     xidVal->Size = xid.length();
     xidVal->Direction = adParamInput;
     xidVal->Value = BlobEncoder(xid);
     cmd->Parameters->Append(xidVal);
     cmd->Execute(NULL, NULL, adCmdText | adExecuteNoRecords);
 }

 void
 MessageMapRecordset::removeForQueue(uint64_t queueId)
 {
     std::ostringstream command;
     command << "DELETE FROM " << tableName
             << " WHERE queueId = " << queueId << std::ends;
     _CommandPtr cmd = NULL;

     TESTHR(cmd.CreateInstance(__uuidof(Command)));
     _ConnectionPtr p = *dbConn;
     cmd->ActiveConnection = p;
     cmd->CommandText = command.str().c_str();
     cmd->CommandType = adCmdText;
     cmd->Execute(NULL, NULL, adCmdText | adExecuteNoRecords);
 }

 void
 MessageMapRecordset::commitPrepared(const std::string& xid)
 {
     // Find all the records for the specified xid. Records marked as adding
     // are now permanent so remove the xid and prepareStatus. Records marked
     // as removing are removed entirely.
     openRs();
     MessageMap m;
     IADORecordBinding *piAdoRecordBinding;
     rs->QueryInterface(__uuidof(IADORecordBinding),
                        (LPVOID *)&piAdoRecordBinding);
     piAdoRecordBinding->BindToRecordset(&m);
     for (; !rs->EndOfFile; rs->MoveNext()) {
         if (m.xidStatus != adFldOK)
             continue;
         const std::string x(m.xid, m.xidLength);
         if (x != xid)
             continue;
         if (m.prepareStatus == PREPARE_REMOVE) {
             rs->Delete(adAffectCurrent);
         }
         else {
             _variant_t dbNull;
             dbNull.ChangeType(VT_NULL);
             rs->Fields->GetItem("prepareStatus")->Value = dbNull;
             rs->Fields->GetItem("xid")->Value = dbNull;
         }
         rs->Update();
     }
     piAdoRecordBinding->Release();
 }

 void
 MessageMapRecordset::abortPrepared(const std::string& xid)
 {
     // Find all the records for the specified xid. Records marked as adding
     // need to be removed while records marked as removing are put back to
     // no xid and no prepareStatus.
     openRs();
     MessageMap m;
     IADORecordBinding *piAdoRecordBinding;
     rs->QueryInterface(__uuidof(IADORecordBinding),
                        (LPVOID *)&piAdoRecordBinding);
     piAdoRecordBinding->BindToRecordset(&m);
     for (; !rs->EndOfFile; rs->MoveNext()) {
         if (m.xidStatus != adFldOK)
             continue;
         const std::string x(m.xid, m.xidLength);
         if (x != xid)
             continue;
         if (m.prepareStatus == PREPARE_ADD) {
             rs->Delete(adAffectCurrent);
         }
         else {
             _variant_t dbNull;
             dbNull.ChangeType(VT_NULL);
             rs->Fields->GetItem("prepareStatus")->Value = dbNull;
             rs->Fields->GetItem("xid")->Value = dbNull;
         }
         rs->Update();
     }
     piAdoRecordBinding->Release();
 }

 void
 MessageMapRecordset::recover(MessageQueueMap& msgMap)
 {
     openRs();
     if (rs->BOF && rs->EndOfFile)
         return;   // Nothing to do
     rs->MoveFirst();
     MessageMap b;
     IADORecordBinding *piAdoRecordBinding;
     rs->QueryInterface(__uuidof(IADORecordBinding),
                        (LPVOID *)&piAdoRecordBinding);
     piAdoRecordBinding->BindToRecordset(&b);
     while (!rs->EndOfFile) {
         qpid::store::QueueEntry entry(b.queueId);
         if (b.xidStatus == adFldOK && b.xidLength > 0) {
             entry.xid.assign(b.xid, b.xidLength);
             entry.tplStatus =
                 b.prepareStatus == PREPARE_ADD ? QueueEntry::ADDING
                                                : QueueEntry::REMOVING;
         }
         else {
             entry.tplStatus = QueueEntry::NONE;
         }
         msgMap[b.messageId].push_back(entry);
         rs->MoveNext();
     }

     piAdoRecordBinding->Release();
 }

 void
 MessageMapRecordset::dump()
 {
     openRs();
     Recordset::dump();
     if (rs->EndOfFile && rs->BOF)    // No records
         return;
     rs->MoveFirst();

     MessageMap m;
     IADORecordBinding *piAdoRecordBinding;
     rs->QueryInterface(__uuidof(IADORecordBinding),
                        (LPVOID *)&piAdoRecordBinding);
     piAdoRecordBinding->BindToRecordset(&m);

     while (!rs->EndOfFile) {
         QPID_LOG(notice, "msg " << m.messageId << " on queue " << m.queueId);
         rs->MoveNext();
     }

     piAdoRecordBinding->Release();
 }

 }}}  // namespace qpid::store::ms_sql
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/

	#include <qpid/Exception.h>
	#include <qpid/log/Statement.h>
	#include <qpid/store/StorageProvider.h>

	#include "MessageMapRecordset.h"
	#include "BlobEncoder.h"
	#include "DatabaseConnection.h"
	#include "Exception.h"
	#include "VariantHelper.h"

	namespace {
	inline void TESTHR(HRESULT x) {if FAILED(x) _com_issue_error(x);};
	}

	namespace qpid {
	namespace store {
	namespace ms_sql {

	void
	MessageMapRecordset::open(DatabaseConnection* conn, const std::string& table)
	{
	init(conn, table);
	}

	void
	MessageMapRecordset::add(uint64_t messageId,
	uint64_t queueId,
	const std::string& xid)
	{
	std::ostringstream command;
	command << "INSERT INTO " << tableName
	<< " (messageId, queueId";
	if (!xid.empty())
	command << ", prepareStatus, xid";
	command << ") VALUES (" << messageId << "," << queueId;
	if (!xid.empty())
	command << "," << PREPARE_ADD << ",?";
	command << ")" << std::ends;

	_CommandPtr cmd = NULL;
	_ParameterPtr xidVal = NULL;
	TESTHR(cmd.CreateInstance(__uuidof(Command)));
	_ConnectionPtr p = *dbConn;
	cmd->ActiveConnection = p;
	cmd->CommandText = command.str().c_str();
	cmd->CommandType = adCmdText;
	if (!xid.empty()) {
	TESTHR(xidVal.CreateInstance(__uuidof(Parameter)));
	xidVal->Name = "@xid";
	xidVal->Type = adVarBinary;
	xidVal->Size = xid.length();
	xidVal->Direction = adParamInput;
	xidVal->Value = BlobEncoder(xid);
	cmd->Parameters->Append(xidVal);
	}
	cmd->Execute(NULL, NULL, adCmdText \| adExecuteNoRecords);
	}

	void
	MessageMapRecordset::remove(uint64_t messageId, uint64_t queueId)
	{
	std::ostringstream command;
	command << "DELETE FROM " << tableName
	<< " WHERE queueId = " << queueId
	<< " AND messageId = " << messageId << std::ends;
	_CommandPtr cmd = NULL;
	TESTHR(cmd.CreateInstance(__uuidof(Command)));
	_ConnectionPtr p = *dbConn;
	cmd->ActiveConnection = p;
	cmd->CommandText = command.str().c_str();
	cmd->CommandType = adCmdText;
	_variant_t deletedRecords;
	cmd->Execute(&deletedRecords, NULL, adCmdText \| adExecuteNoRecords);
	if ((long)deletedRecords == 0)
	throw ms_sql::Exception("Message does not exist in queue mapping");
	// Trigger on deleting the mapping takes care of deleting orphaned
	// message record from tblMessage.
	}

	void
	MessageMapRecordset::pendingRemove(uint64_t messageId,
	uint64_t queueId,
	const std::string& xid)
	{
	// Look up the mapping for the specified message and queue. There
	// should be only one because of the uniqueness constraint in the
	// SQL table. Update it to reflect it's pending delete with
	// the specified xid.
	std::ostringstream command;
	command << "UPDATE " << tableName
	<< " SET prepareStatus=" << PREPARE_REMOVE
	<< " , xid=?"
	<< " WHERE queueId = " << queueId
	<< " AND messageId = " << messageId << std::ends;

	_CommandPtr cmd = NULL;
	_ParameterPtr xidVal = NULL;
	TESTHR(cmd.CreateInstance(__uuidof(Command)));
	TESTHR(xidVal.CreateInstance(__uuidof(Parameter)));
	_ConnectionPtr p = *dbConn;
	cmd->ActiveConnection = p;
	cmd->CommandText = command.str().c_str();
	cmd->CommandType = adCmdText;
	xidVal->Name = "@xid";
	xidVal->Type = adVarBinary;
	xidVal->Size = xid.length();
	xidVal->Direction = adParamInput;
	xidVal->Value = BlobEncoder(xid);
	cmd->Parameters->Append(xidVal);
	cmd->Execute(NULL, NULL, adCmdText \| adExecuteNoRecords);
	}

	void
	MessageMapRecordset::removeForQueue(uint64_t queueId)
	{
	std::ostringstream command;
	command << "DELETE FROM " << tableName
	<< " WHERE queueId = " << queueId << std::ends;
	_CommandPtr cmd = NULL;

	TESTHR(cmd.CreateInstance(__uuidof(Command)));
	_ConnectionPtr p = *dbConn;
	cmd->ActiveConnection = p;
	cmd->CommandText = command.str().c_str();
	cmd->CommandType = adCmdText;
	cmd->Execute(NULL, NULL, adCmdText \| adExecuteNoRecords);
	}

	void
	MessageMapRecordset::commitPrepared(const std::string& xid)
	{
	// Find all the records for the specified xid. Records marked as adding
	// are now permanent so remove the xid and prepareStatus. Records marked
	// as removing are removed entirely.
	openRs();
	MessageMap m;
	IADORecordBinding *piAdoRecordBinding;
	rs->QueryInterface(__uuidof(IADORecordBinding),
	(LPVOID *)&piAdoRecordBinding);
	piAdoRecordBinding->BindToRecordset(&m);
	for (; !rs->EndOfFile; rs->MoveNext()) {
	if (m.xidStatus != adFldOK)
	continue;
	const std::string x(m.xid, m.xidLength);
	if (x != xid)
	continue;
	if (m.prepareStatus == PREPARE_REMOVE) {
	rs->Delete(adAffectCurrent);
	}
	else {
	_variant_t dbNull;
	dbNull.ChangeType(VT_NULL);
	rs->Fields->GetItem("prepareStatus")->Value = dbNull;
	rs->Fields->GetItem("xid")->Value = dbNull;
	}
	rs->Update();
	}
	piAdoRecordBinding->Release();
	}

	void
	MessageMapRecordset::abortPrepared(const std::string& xid)
	{
	// Find all the records for the specified xid. Records marked as adding
	// need to be removed while records marked as removing are put back to
	// no xid and no prepareStatus.
	openRs();
	MessageMap m;
	IADORecordBinding *piAdoRecordBinding;
	rs->QueryInterface(__uuidof(IADORecordBinding),
	(LPVOID *)&piAdoRecordBinding);
	piAdoRecordBinding->BindToRecordset(&m);
	for (; !rs->EndOfFile; rs->MoveNext()) {
	if (m.xidStatus != adFldOK)
	continue;
	const std::string x(m.xid, m.xidLength);
	if (x != xid)
	continue;
	if (m.prepareStatus == PREPARE_ADD) {
	rs->Delete(adAffectCurrent);
	}
	else {
	_variant_t dbNull;
	dbNull.ChangeType(VT_NULL);
	rs->Fields->GetItem("prepareStatus")->Value = dbNull;
	rs->Fields->GetItem("xid")->Value = dbNull;
	}
	rs->Update();
	}
	piAdoRecordBinding->Release();
	}

	void
	MessageMapRecordset::recover(MessageQueueMap& msgMap)
	{
	openRs();
	if (rs->BOF && rs->EndOfFile)
	return; // Nothing to do
	rs->MoveFirst();
	MessageMap b;
	IADORecordBinding *piAdoRecordBinding;
	rs->QueryInterface(__uuidof(IADORecordBinding),
	(LPVOID *)&piAdoRecordBinding);
	piAdoRecordBinding->BindToRecordset(&b);
	while (!rs->EndOfFile) {
	qpid::store::QueueEntry entry(b.queueId);
	if (b.xidStatus == adFldOK && b.xidLength > 0) {
	entry.xid.assign(b.xid, b.xidLength);
	entry.tplStatus =
	b.prepareStatus == PREPARE_ADD ? QueueEntry::ADDING
	: QueueEntry::REMOVING;
	}
	else {
	entry.tplStatus = QueueEntry::NONE;
	}
	msgMap[b.messageId].push_back(entry);
	rs->MoveNext();
	}

	piAdoRecordBinding->Release();
	}

	void
	MessageMapRecordset::dump()
	{
	openRs();
	Recordset::dump();
	if (rs->EndOfFile && rs->BOF) // No records
	return;
	rs->MoveFirst();

	MessageMap m;
	IADORecordBinding *piAdoRecordBinding;
	rs->QueryInterface(__uuidof(IADORecordBinding),
	(LPVOID *)&piAdoRecordBinding);
	piAdoRecordBinding->BindToRecordset(&m);

	while (!rs->EndOfFile) {
	QPID_LOG(notice, "msg " << m.messageId << " on queue " << m.queueId);
	rs->MoveNext();
	}

	piAdoRecordBinding->Release();
	}

	}}} // namespace qpid::store::ms_sql