#include <NdbApi.hpp>
#include <stdio.h>
#include <iostream>
#include <unistd.h>
#ifdef VM_TRACE
#include <my_global.h>
#endif
#ifndef assert
#include <assert.h>
#endif
#define APIERROR(error) \
{ std::cout << "Error in " << __FILE__ << ", line:" << __LINE__ << ", code:" \
<< error.code << ", msg: " << error.message << "." << std::endl; \
exit(-1); }
int myCreateEvent(Ndb* myNdb,
const char *eventName,
const char *eventTableName,
const char **eventColumnName,
const int noEventColumnName,
bool merge_events);
int main(int argc, char** argv)
{
if (argc < 3)
{
std::cout << "Arguments are <connect_string cluster> <timeout> [m(merge events)|d(debug)].\n";
exit(-1);
}
const char *connectstring = argv[1];
int timeout = atoi(argv[2]);
ndb_init();
bool merge_events = argc > 3 && strchr(argv[3], 'm') != 0;
#ifdef VM_TRACE
bool dbug = argc > 3 && strchr(argv[3], 'd') != 0;
if (dbug) DBUG_PUSH("d:t:");
if (dbug) putenv("API_SIGNAL_LOG=-");
#endif
Ndb_cluster_connection *cluster_connection=
new Ndb_cluster_connection(connectstring);
int r= cluster_connection->connect(5 ,
3 ,
1 );
if (r > 0)
{
std::cout
<< "Cluster connect failed, possibly resolved with more retries.\n";
exit(-1);
}
else if (r < 0)
{
std::cout
<< "Cluster connect failed.\n";
exit(-1);
}
if (cluster_connection->wait_until_ready(30,30))
{
std::cout << "Cluster was not ready within 30 secs." << std::endl;
exit(-1);
}
Ndb* myNdb= new Ndb(cluster_connection,
"TEST_DB");
if (myNdb->init() == -1) APIERROR(myNdb->getNdbError());
const char *eventName= "CHNG_IN_t0";
const char *eventTableName= "t0";
const int noEventColumnName= 5;
const char *eventColumnName[noEventColumnName]=
{"c0",
"c1",
"c2",
"c3",
"c4"
};
myCreateEvent(myNdb,
eventName,
eventTableName,
eventColumnName,
noEventColumnName,
merge_events);
typedef union { NdbRecAttr* ra; NdbBlob* bh; } RA_BH;
int i, j, k, l;
j = 0;
while (j < timeout) {
NdbEventOperation* op;
printf("create EventOperation\n");
if ((op = myNdb->createEventOperation(eventName)) == NULL)
APIERROR(myNdb->getNdbError());
op->mergeEvents(merge_events);
printf("get values\n");
RA_BH recAttr[noEventColumnName];
RA_BH recAttrPre[noEventColumnName];
for (i = 0; i < noEventColumnName; i++) {
if (i < 4) {
recAttr[i].ra = op->getValue(eventColumnName[i]);
recAttrPre[i].ra = op->getPreValue(eventColumnName[i]);
} else if (merge_events) {
recAttr[i].bh = op->getBlobHandle(eventColumnName[i]);
recAttrPre[i].bh = op->getPreBlobHandle(eventColumnName[i]);
}
}
printf("execute\n");
if (op->execute())
APIERROR(op->getNdbError());
NdbEventOperation* the_op = op;
i= 0;
while (i < timeout) {
int r = myNdb->pollEvents(1000);
if (r > 0) {
while ((op= myNdb->nextEvent())) {
assert(the_op == op);
i++;
switch (op->getEventType()) {
case NdbDictionary::Event::TE_INSERT:
printf("%u INSERT", i);
break;
case NdbDictionary::Event::TE_DELETE:
printf("%u DELETE", i);
break;
case NdbDictionary::Event::TE_UPDATE:
printf("%u UPDATE", i);
break;
default:
abort();
}
printf(" gci=%d\n", (int)op->getGCI());
for (k = 0; k <= 1; k++) {
printf(k == 0 ? "post: " : "pre : ");
for (l = 0; l < noEventColumnName; l++) {
if (l < 4) {
NdbRecAttr* ra = k == 0 ? recAttr[l].ra : recAttrPre[l].ra;
if (ra->isNULL() >= 0) {
if (ra->isNULL() == 0) {
if (l < 2)
printf("%-5u", ra->u_32_value());
else
printf("%-5.4s", ra->aRef());
} else
printf("%-5s", "NULL");
} else
printf("%-5s", "-");
} else if (merge_events) {
int isNull;
NdbBlob* bh = k == 0 ? recAttr[l].bh : recAttrPre[l].bh;
bh->getDefined(isNull);
if (isNull >= 0) {
if (! isNull) {
Uint64 length = 0;
bh->getLength(length);
unsigned char* buf = new unsigned char [length];
memset(buf, 'X', length);
Uint32 n = length;
bh->readData(buf, n);
assert(n == length);
bool first = true;
Uint32 i = 0;
while (i < n) {
unsigned char c = buf[i++];
Uint32 m = 1;
while (i < n && buf[i] == c)
i++, m++;
if (! first)
printf("+");
printf("%u%c", m, c);
first = false;
}
printf("[%u]", n);
delete [] buf;
} else
printf("%-5s", "NULL");
} else
printf("%-5s", "-");
}
}
printf("\n");
}
}
} else
printf("timed out (%i)\n", timeout);
}
if (myNdb->dropEventOperation(the_op)) APIERROR(myNdb->getNdbError());
the_op = 0;
j++;
}
{
NdbDictionary::Dictionary *myDict = myNdb->getDictionary();
if (!myDict) APIERROR(myNdb->getNdbError());
if (myDict->dropEvent(eventName)) APIERROR(myDict->getNdbError());
}
delete myNdb;
delete cluster_connection;
ndb_end(0);
return 0;
}
int myCreateEvent(Ndb* myNdb,
const char *eventName,
const char *eventTableName,
const char **eventColumnNames,
const int noEventColumnNames,
bool merge_events)
{
NdbDictionary::Dictionary *myDict= myNdb->getDictionary();
if (!myDict) APIERROR(myNdb->getNdbError());
const NdbDictionary::Table *table= myDict->getTable(eventTableName);
if (!table) APIERROR(myDict->getNdbError());
NdbDictionary::Event myEvent(eventName, *table);
myEvent.addTableEvent(NdbDictionary::Event::TE_ALL);
myEvent.addEventColumns(noEventColumnNames, eventColumnNames);
myEvent.mergeEvents(merge_events);
if (myDict->createEvent(myEvent) == 0)
myEvent.print();
else if (myDict->getNdbError().classification ==
NdbError::SchemaObjectExists) {
printf("Event creation failed, event exists\n");
printf("dropping Event...\n");
if (myDict->dropEvent(eventName)) APIERROR(myDict->getNdbError());
if ( myDict->createEvent(myEvent)) APIERROR(myDict->getNdbError());
} else
APIERROR(myDict->getNdbError());
return 0;
}