/*
* engine_client.cpp
*
* This simple example illustrates how to use the C++ SDK to
* interface with Apama. The example connects to a remote Event Correlator
* (also called a Correlation Engine or just the Engine), creates
* an event consumer and connects it to the Engine's supplier interface,
* creates some MonitorScript code and injects it, injects some sample events,
* and receives some back from the Engine when the monitor triggers. It then
* disconnects from the Engine and exits.
*
* This sample uses the C++ SDK. It has only been tested against Sun's
* CC 5.5 on Solaris or GCC 3.0.4 on Linux. Other compilers may
* generate code that will not link with the Apama runtime library.
*
* Copyright(c) 2002, 2004-2005 Software AG. All rights
* reserved. Use, reproduction, transfer, publication or disclosure is
* prohibited except as specifically provided for in your License Agreement
* with PSC.
*
* $RCSfile: engine_client.cpp,v $ $Revision: 1.5.6.1 $ $Date: 2006/04/03 12:31:20 $
*/
 
#include <engine_client_cpp.hpp>
#include <iostream>
#include <stddef.h>
#include <stdlib.h>
#ifdef __unix__
#include <unistd.h>
#endif
#ifdef __WIN32__
#include <windows.h>
#endif
 
using namespace std;
 
using com::apama::engine::EngineManagement;
using com::apama::engine::EngineStatus;
using com::apama::engine::MonitorScript;
using com::apama::event::EventConsumer;
using com::apama::event::EventSupplier;
using com::apama::event::Event;
using com::apama::EngineException;
 
/**
* Event receiver implementation.
*/
class receive_consumer : public EventConsumer {
 
public:
/**
* Constructor creates the connection to the Engine.
*
* @param engine The Engine to connect to.
*
* @exception EngineException
*/
receive_consumer(EngineManagement* engine);
 
/**
* Destructor disconnects from Engine.
*/
virtual ~receive_consumer();
 
/**
* Receive events from the Engine and log them to stdout.
*
* @param events The received events.
*
* @exception EngineException
*/
virtual void sendEvents(const Event* const * events);
 
private:
/** The Engine we're connected to */
EngineManagement* engine;
 
/** Per-connection resource handle returned by the Engine */
EventSupplier* supplier;
};
 
/**
* Create connection to engine.
*/
receive_consumer::receive_consumer(EngineManagement* eng) : EventConsumer(),
engine(eng), supplier(NULL) {
// Make the connection. The returned EventSupplier is a handle to
// private resources allocated by the Engine to deal with this
// connection. These will be freed by calling
// deleteEventSupplier() in the destructor for this class.
supplier = engine->connectEventConsumer(this, NULL);
}
 
/**
* Disconnect from Engine
*/
receive_consumer::~receive_consumer() {
// Disconnect from the Engine and free per-connection resources
com::apama::event::deleteEventSupplier(supplier);
}
 
/**
* Receive a batch of events, log to stdout.
*/
void receive_consumer::sendEvents(const Event* const * events) {
for (const Event* const * event = events; *event; event++) {
cout << **event << endl;
}
}
 
/**
* Main program.
*
* Return codes:
* 0 = Everything OK
* 1 = Couldn't connect to Engine
* 2 = Something else went wrong
*/
int main(int argc, const char** argv) {
// Return code
int rc = 2;
 
// Error message to display if anything goes wrong. Update this
// appropriately before each operation that might break.
const char* emsg;
 
if (argc == 3 && atoi(argv[2]) > 0) {
 
// Set to true once the Engine library has been initialised
bool initDone = false;
 
// The engine
EngineManagement* engine = NULL;
 
try {
try {
// Initialise Engine client-side library
rc = 1;
emsg = "Failed to initialise engine library";
com::apama::engine::engineInit();
initDone = true;
 
// Attempt to connect to a remote Engine
emsg = "Failed to connect to engine";
engine = com::apama::engine::connectToEngine(argv[1], atoi(argv[2]));
 
// Create an event consumer
emsg = "Event sink connection failed";
receive_consumer* consumer = new receive_consumer(engine);
 
// Inject some MonitorScript (don't forget to delete it when done)
emsg = "MonitorScript injection failed";
MonitorScript* script = com::apama::engine::createMonitorScript(
"event TestEvent {"
"string text;"
"}"
""
"monitor Echo {"
""
"TestEvent test;"
""
"action onload {"
"on all TestEvent(*):test {"
"emit TestEvent(test.text);"
"}"
"}"
"}");
engine->injectMonitorScript(*script);
com::apama::engine::deleteMonitorScript(script);
 
// Wait a few seconds to be sure the injection event has been processed
#ifdef __unix__
sleep(3);
#endif
#ifdef __WIN32__
Sleep(3000);
#endif
// Get & display status (have to delete it when done)
emsg = "Status gathering failed";
EngineStatus* status = engine->getStatus();
cout << *status << endl;
com::apama::engine::deleteStatus(status);
 
// Send some events (again, remember to delete Event objects when done)
emsg = "Event sending failed";
Event* events[3];
events[0] = com::apama::event::createEvent("TestEvent(\"Hello, World\")");
events[1] = com::apama::event::createEvent("TestEvent(\"Welcome to Apama\")");
events[2] = NULL;
engine->sendEvents(events);
com::apama::event::deleteEvent(events[0]);
com::apama::event::deleteEvent(events[1]);
 
// Delete the event type and monitor we added
emsg = "Name deletion failed";
engine->deleteName("Echo");
engine->deleteName("TestEvent");
 
// Wait a few seconds for the output event to be received
// and the deletions processed
#ifdef __unix__
sleep(3);
#endif
#ifdef __WIN32__
Sleep(3000);
#endif
 
// Display status again
emsg = "Status gathering failed";
cout << endl;
status = engine->getStatus();
cout << *status << endl;
com::apama::engine::deleteStatus(status);
 
// Disconnect and destroy the event consumer
emsg = "Event sink disconnection failed";
delete consumer;
 
// If we got this far, everything succeeded!
rc = 0;
}
catch (EngineException& ex) {
// Rethrow so exception printer can deal with it
throw ex;
}
catch (...) {
throw EngineException("Caught non-engine exception in main()");
}
}
catch (EngineException& ex) {
cerr << emsg << ": " << ex.what() << endl;
}
 
try {
try {
// Shutdown cleanly.
if (engine) {
// Disconnect from the Engine
emsg = "Failed to disconnect from engine";
com::apama::engine::disconnectFromEngine(engine);
}
if (initDone) {
// Shutdown the Engine library
emsg = "Failed to shutdown engine library";
com::apama::engine::engineShutdown();
}
}
catch (EngineException& ex) {
// Rethrow so exception printer can deal with it
throw ex;
}
catch (...) {
throw EngineException("Caught non-engine exception in main()");
}
}
catch (EngineException& ex) {
cerr << emsg << ": " << ex.what() << endl;
}
}
else {
// Bad command line given
cerr << "Usage: " << argv[0] << " <host> <port>" << endl;
}
 
// Done!
return rc;
}