/*
* engine_client.c
*
* This simple example illustrates how to use the C SDK to
* interface with Apama. The example connects to a remote Event Correlator
* (also known as a Correlation Engine), creates an event consumer and connects
* it to Apama'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.
*
* 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.c,v $ $Revision: 1.5.6.1 $ $Date: 2006/04/03 12:31:20 $
*/
#include <engine_client_c.h>
#include <stddef.h>
#include <stdlib.h>
#ifdef __unix__
#include <unistd.h>
#endif
#ifdef __WIN32__
#include <windows.h>
#endif
#include <stdio.h>
#define STATUS_BUFFER_SIZE 8192
/**
* Receive a batch of events, log to stdout.
*/
static void AP_ENGINE_CLIENT_CALL receiveConsumer_sendEvents(
AP_EventConsumer* consumer, AP_Event** events) {
AP_Event** event;
for (event = events; *event; event++) {
printf("%s\n", (*event)->functions->getText(*event));
}
}
/**
* Function table for our event consumer.
*/
static struct AP_EventConsumer_Functions receiveConsumer_Functions = {
receiveConsumer_sendEvents
};
/**
* 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;
/* Buffer for stringified status reports */
AP_char8 statusBuf[STATUS_BUFFER_SIZE];
if (argc == 3 && atoi(argv[2]) > 0) {
/* Set to true once the engine library has been initialised */
AP_bool initDone = 0;
/* Set to true to drop out of the loop with a pending exception */
AP_bool exThrown = 0;
/* The engine */
AP_EngineManagement* engine = NULL;
do {
/* The event consumer */
AP_EventConsumer* consumer = NULL;
/* The (remote) event supplier reference */
AP_EventSupplier* supplier = NULL;
/* Monitorscript to be injected into engine */
AP_MonitorScript* script = NULL;
/* Engine status report */
AP_EngineStatus* status = NULL;
/* Events to be sent */
AP_Event* events[3];
/* Initialise Apama SDK client-side library */
rc = 1;
emsg = "Failed to initialise Apama SDK library";
AP_EngineInit();
if (AP_CheckException()) {
exThrown = 1;
break;
}
initDone = 1;
/* Attempt to connect to a remote Engine */
emsg = "Failed to connect to engine";
engine = AP_ConnectToEngine(argv[1], (AP_uint16) atoi(argv[2]));
if (AP_CheckException() || !engine) {
exThrown = 1;
break;
}
/* Create an event consumer */
emsg = "Event sink connection failed";
consumer = (AP_EventConsumer*)malloc(sizeof(AP_EventConsumer));
consumer->functions = &receiveConsumer_Functions;
supplier = engine->functions->connectEventConsumer(engine, consumer, NULL);
if (AP_CheckException() || !supplier) {
exThrown = 1;
break;
}
/* Inject some MonitorScript (don't forget to delete it when done) */
emsg = "MonitorScript injection failed";
script = AP_CreateMonitorScript(
"event TestEvent {"
"string text;"
"}"
""
"monitor Echo {"
""
"TestEvent test;"
""
"action onload {"
"on all TestEvent(*):test {"
"emit TestEvent(test.text);"
"}"
"}"
"}");
engine->functions->injectMonitorScript(engine, script);
if (AP_CheckException()) {
exThrown = 1;
break;
}
AP_DeleteMonitorScript(script);
if (AP_CheckException()) {
exThrown = 1;
break;
}
/* 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";
printf("\n");
status = engine->functions->getStatus(engine);
if (AP_CheckException() || !status) {
exThrown = 1;
break;
}
if (status->functions->print(status, statusBuf, STATUS_BUFFER_SIZE)) {
printf("%s", statusBuf);
}
else {
printf("Status buffer too small!");
}
printf("\n");
AP_DeleteEngineStatus(status);
/* Send some events (again, remember to delete Event objects when done) */
emsg = "Event sending failed";
events[0] = AP_CreateEvent("TestEvent(\"Hello, World\")");
events[1] = AP_CreateEvent("TestEvent(\"Welcome to Apama\")");
events[2] = NULL;
engine->s_EventConsumer->functions->sendEvents(engine->s_EventConsumer, events);
AP_DeleteEvent(events[0]);
AP_DeleteEvent(events[1]);
if (AP_CheckException()) {
exThrown = 1;
break;
}
/* Delete the event type and monitor we added */
emsg = "Name deletion failed";
engine->functions->deleteName(engine, "Echo");
if (AP_CheckException()) {
exThrown = 1;
break;
}
engine->functions->deleteName(engine, "TestEvent");
if (AP_CheckException()) {
exThrown = 1;
break;
}
/* 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";
printf("\n");
status = engine->functions->getStatus(engine);
if (AP_CheckException() || !status) {
exThrown = 1;
break;
}
if (status->functions->print(status, statusBuf, STATUS_BUFFER_SIZE)) {
printf("%s", statusBuf);
}
else {
printf("Status buffer too small!");
}
printf("\n");
AP_DeleteEngineStatus(status);
/* Disconnect and destroy the event consumer */
emsg = "Event sink disconnection failed";
supplier->functions->disconnect(supplier);
if (AP_CheckException()) {
exThrown = 1;
break;
}
free((void*)consumer);
/* If we got this far, everything succeeded! */
rc = 0;
} while (0);
if (exThrown) {
if (AP_CheckException()) {
printf("%s: %s\n", emsg, AP_GetExceptionMessage());
}
else {
printf("%s\n", emsg);
}
exThrown = 0;
AP_ClearException();
}
do {
/* Shutdown cleanly */
if (engine) {
/* Disconnect from the engine */
emsg = "Failed to disconnect from Engine";
AP_DisconnectFromEngine(engine);
if (AP_CheckException()) {
exThrown = 1;
}
}
if (initDone) {
/* Shutdown the engine library */
emsg = "Failed to shutdown Apama SDK library";
AP_EngineShutdown();
if (AP_CheckException()) {
exThrown = 1;
}
}
} while (0);
if (exThrown) {
if (AP_CheckException()) {
printf("%s: %s\n", emsg, AP_GetExceptionMessage());
}
else {
printf("%s\n", emsg);
}
exThrown = 0;
AP_ClearException();
}
}
else {
/* Bad command line given */
fprintf(stderr, "Usage: %s <host> <port>\n", argv[0]);
}
/* Done! */
return rc;
}