| /* |
| * The Continuous Query QuickStart Example. |
| * |
| * This example takes the following steps: |
| * |
| * 1. Create a GemFire Cache Programmatically. |
| * 2. Create the example Region Programmatically. |
| * 3. Populate some query objects on the Region. |
| * 4. Register a cqQuery listener |
| * 5. Execute a cqQuery with initial Results |
| * 6. Close the Cache. |
| * |
| */ |
| |
| // Include the GemFire library. |
| #include <gfcpp/GemfireCppCache.hpp> |
| |
| // Include our Query objects, viz. Portfolio and Position. |
| #include "queryobjects/Portfolio.hpp" |
| #include "queryobjects/Position.hpp" |
| |
| // Use the "gemfire" namespace. |
| using namespace gemfire; |
| |
| // Use the "testobject" namespace for the query objects. |
| using namespace testobject; |
| |
| class MyCqListener : public CqListener { |
| public: |
| void onEvent(const CqEvent& cqe) |
| { |
| char* opStr = (char*)"Default"; |
| PortfolioPtr portfolio( dynamic_cast<Portfolio*> (cqe.getNewValue().ptr() )); |
| CacheableStringPtr key( dynamic_cast<CacheableString*> (cqe.getKey().ptr() )); |
| switch (cqe.getQueryOperation()) |
| { |
| case CqOperation::OP_TYPE_CREATE: |
| { |
| opStr = (char*)"CREATE"; |
| break; |
| } |
| case CqOperation::OP_TYPE_UPDATE: |
| { |
| opStr = (char*)"UPDATE"; |
| break; |
| } |
| case CqOperation::OP_TYPE_DESTROY: |
| { |
| opStr = (char*)"UPDATE"; |
| break; |
| } |
| default: |
| break; |
| } |
| LOGINFO("MyCqListener::OnEvent called with %s, key[%s], value=(%ld,%s)", |
| opStr, key->asChar(), portfolio->getID(), portfolio->getPkid()->asChar()); |
| } |
| |
| void onError(const CqEvent& cqe){ |
| LOGINFO("MyCqListener::OnError called"); |
| } |
| |
| void close(){ |
| LOGINFO("MyCqListener::close called"); |
| } |
| }; |
| |
| // The CqQuery QuickStart example. |
| int main(int argc, char ** argv) |
| { |
| try |
| { |
| // Create a GemFire Cache Programmatically. |
| CacheFactoryPtr cacheFactory = CacheFactory::createCacheFactory(); |
| |
| CachePtr cachePtr = cacheFactory->addServer("localhost", 50505) |
| ->setSubscriptionEnabled(true) |
| ->create(); |
| |
| LOGINFO("Created the GemFire Cache"); |
| |
| RegionFactoryPtr regionFactory = cachePtr->createRegionFactory(PROXY); |
| |
| LOGINFO("Created the RegionFactory"); |
| |
| // Create the example Region programmatically. |
| RegionPtr regionPtr = regionFactory->create("Portfolios"); |
| |
| // Register our Serializable/Cacheable Query objects, viz. Portfolio and Position. |
| Serializable::registerType( Portfolio::createDeserializable); |
| Serializable::registerType( Position::createDeserializable); |
| |
| LOGINFO("Registered Serializable Query Objects"); |
| |
| // Populate the Region with some Portfolio objects. |
| PortfolioPtr port1Ptr(new Portfolio(1 /*ID*/, 10 /*size*/)); |
| PortfolioPtr port2Ptr(new Portfolio(2 /*ID*/, 20 /*size*/)); |
| PortfolioPtr port3Ptr(new Portfolio(3 /*ID*/, 30 /*size*/)); |
| regionPtr->put("Key1", port1Ptr); |
| regionPtr->put("Key2", port2Ptr); |
| regionPtr->put("Key3", port3Ptr); |
| |
| LOGINFO("Populated some Portfolio Objects"); |
| |
| // Get the QueryService from the Cache. |
| QueryServicePtr qrySvcPtr = cachePtr->getQueryService(); |
| |
| //Create CqAttributes and Install Listener |
| CqAttributesFactory cqFac; |
| CqListenerPtr cqLstner (new MyCqListener()); |
| cqFac.addCqListener(cqLstner); |
| CqAttributesPtr cqAttr = cqFac.create(); |
| |
| //create a new Cq Query |
| const char* qryStr = "select * from /Portfolios p where p.ID < 5"; |
| CqQueryPtr qry = qrySvcPtr->newCq((char*)"MyCq", qryStr, cqAttr); |
| |
| //execute Cq Query with initial Results |
| CqResultsPtr resultsPtr = qry->executeWithInitialResults(); |
| |
| //make change to generate cq events |
| regionPtr->put("Key3", port1Ptr); |
| regionPtr->put("Key2", port2Ptr); |
| regionPtr->put("Key1", port3Ptr); |
| |
| LOGINFO("ResultSet Query returned %d rows", resultsPtr->size()); |
| |
| // Iterate through the rows of the query result. |
| SelectResultsIterator iter = resultsPtr->getIterator(); |
| while (iter.hasNext()) |
| { |
| SerializablePtr ser = iter.next(); |
| if (ser != NULLPTR) { |
| LOGINFO (" query pulled object %s\n", ser->toString()->asChar()); |
| |
| StructPtr stPtr(dynamic_cast<Struct*> (ser.ptr() )); |
| if (stPtr != NULLPTR) |
| { |
| LOGINFO(" got struct ptr "); |
| SerializablePtr serKey = (*(stPtr.ptr()))["key"]; |
| if (serKey != NULLPTR) |
| { |
| LOGINFO("got struct key %s\n", serKey->toString()->asChar()); |
| } |
| |
| SerializablePtr serVal = (*(stPtr.ptr()))["value"]; |
| |
| if (serVal != NULLPTR) |
| { |
| LOGINFO(" got struct value %s\n", serVal->toString()->asChar()); |
| } |
| } |
| } |
| else { |
| printf("query pulled bad object\n"); |
| } |
| } |
| // Stop the GemFire Continuous query. |
| qry->stop(); |
| // Close the GemFire Continuous query. |
| qry->close(); |
| |
| // Close the GemFire Cache. |
| cachePtr->close(); |
| |
| LOGINFO("Closed the GemFire Cache"); |
| |
| } |
| // An exception should not occur |
| catch(const Exception & gemfireExcp) |
| { |
| LOGERROR("CqQuery GemFire Exception: %s", gemfireExcp.getMessage()); |
| } |
| } |
| |