4.2 An Example Using the Threads Module and the DB Interface Module

SourcePro DB is designed to work with the other SourcePro C++ products to supply database access. The following example combines class RWDBDatabase, from the DB Interface Module of SourcePro DB, with class RWServerPool, from the Threads Module of SourcePro Core, to create a higher-level abstraction, the multithreaded class QueryServer. In just a few lines of code, this class manages a controlled number of threads that can perform database queries for the whole application.

In the code using QueryServer, the server is constructed with an RWDBDatabase object and the number of threads to be used at one time for database queries is specified. Following that, application code can submit queries to the QueryServer using its enqueue() method. Internally, QueryServer manages its threads using RWServerPool, and asynchronously gives results to users via the RWTIOUResult template class. When the application is done with the -QueryServer, its destructor cleans up all the worker threads it created, and automatically frees up any used connections to the database.

Example 3: Using the Threads Module and the DB Interface Module

// Header file for class QueryServer // // From the DB Interface Module #include <rw/db/db.h> // From the Threads Module #include <rw/itc/RWTIOUResult.h> #include <rw/thread/RWTRunnableIOUFunction.h> #include <rw/thread/rwtMakeRunnableIOUFunction.h> #include <rw/thread/RWServerPool.h> class QueryServer { public: QueryServer(const RWDBDatabase &db, unsigned concurrency) : database_(db), pool_(RWServerPool::make(concurrency)) { // Set the database's connection pool size to our // concurrency level for better performance. database_.defaultConnections(concurrency); // Start the pool running to accept requests. pool_.start(); } ~QueryServer() { // Wait for the pool to finish processing its queue. pool_.stop(); pool_.join(); } RWTIOUResult<RWDBResult> enqueue(const RWCString& sql) { // Make a new runnable to handle the request and // throw it on the queue. RWTRunnableIOUFunction<RWDBResult> rf = rwtMakeRunnableIOUFunctionMA1(QueryServer, // class *this, // instance RWDBResult, // return type &QueryServer::serviceFunction, // function const RWCString&, // first arg type sql); // first arg pool_.enqueue(rf); return rf.result(); } private: RWDBDatabase database_; RWServerPool pool_; // Service function enqueued on the RWServerPool to handle // the request. Private, because users should use enqueue() // to place a query on the queue. RWDBResult serviceFunction(const RWCString& sql) { // Grab a connection, go for it. return database_.connection().executeSql(sql); } // Not defined on purpose QueryServer(const QueryServer&); QueryServer& operator=(const QueryServer&); }; // Implementation file, queryserver.cpp #include <rw/db/db.h> #include <iostream> #include "queryserver.h" #include <rw/thread/RWRunnableSelf.h> // for rwSleep() #include <rw/tools/bustring.h> using namespace std; void outputStatus(const RWDBStatus& aStatus) { if(aStatus.errorCode() == 0) return; // Print out the error. cout << "Error code: " << (int) aStatus.errorCode() << endl << "Error message " << aStatus.message() << endl << "Is terminal: " << (aStatus.isTerminal() ? "Yes" : "No") << endl << "Vendor error 1: " << aStatus.vendorError1() << endl << "Vendor error 2 : " << aStatus.vendorError2() << endl << "Vendor message 1: " << aStatus.vendorMessage1() << endl << "Vendor message 2: " << aStatus.vendorMessage2() << endl; } int main() { RWDBManager::setErrorHandler(outputStatus); RWDBDatabase db = RWDBManager::database("ORACLEOCI", "database", "username", "password", ""); RWDBTracer& trc= db.tracer(); trc.setOn(RWDBTracer::SQL); trc.stream(cout); RWDBConnection conn = db.connection(); // Create and populate table conn.executeSql("create table testServ(col1 nvarchar2(20))"); // Any Unicode String May be inserted here conn.executeSql("insert into testServ values(N'Oregon Ducks')"); // Create QueryServer QueryServer server(db, 5); RWCString query("select * from testServ"); RWTIOUResult<RWDBResult> resultIOU = server.enqueue(query); // Poll until redeemable while(!resultIOU.redeemable()) { cout << "Waiting for result..." << endl; ::rwSleep(50); } // Obtain RWDBResult object from RWTIOUResult RWDBResult res = resultIOU.redeem(); // Create RWDBReader RWDBReader rdr = res.table().reader(); // Output data from query RWBasicUString output; // Create a file to write out the results RWFile fileOut("queryServerOut.txt"); while(rdr()) { rdr >> output; // Output UString fileOut << output; // Print out UTF 8 version of UString cout << output.toUtf8() << endl; } // Cleanup conn.executeSql("drop table testServ"); return 0; }

Program output:

08/20/2002 12:44:27 SQL> create table testServ(col1 nvarchar2(20)) 08/20/2002 12:44:27 SQL> insert into testServ values(N'Oregon Ducks') Waiting for result... Waiting for result... Waiting for result... Waiting for result... Waiting for result... Waiting for result... Waiting for result... Waiting for result... Waiting for result... 08/20/2002 12:44:28 SQL> select * from testServ Oregon Ducks 08/20/2002 12:44:28 SQL> drop table testServ

Copyright © Rogue Wave Software, Inc. All Rights Reserved.

The Rogue Wave name and logo, and SourcePro, are registered trademarks of Rogue Wave Software. All other trademarks are the property of their respective owners.
Provide feedback to Rogue Wave about its documentation.