test/publish.c

   1 /*
   2  * Test program for event service
   3  */
   4
   5 #include <stdio.h>
   6 #include <string.h>
   7 #include <sys/poll.h>
   8 #include <unistd.h>
   9 #include <fcntl.h>
  10 #ifndef OPENAIS_SOLARIS
  11 #include <stdint.h>
  12 #include <getopt.h>
  13 #else
  14 #include <sys/types.h>
  15 #endif
  16 #include <stdlib.h>
  17 #include <sys/time.h>
  18 #include "saAis.h"
  19 #include "saEvt.h"
  20
  21 // #define EVENT_SUBSCRIBE
  22
  23 #define PUB_RETRIES 100
  24 #define TRY_WAIT 2
  25
  26 extern int get_sa_error(SaAisErrorT, char *, int);
  27 char result_buf[256];
  28 int result_buf_len = sizeof(result_buf);
  29
  30 static int pub_count = 1;
  31 static int wait_time = -1;
  32
  33 SaVersionT version = { 'B', 0x01, 0x01 };
  34
  35 void event_callback( SaEvtSubscriptionIdT subscriptionId,
  36                 const SaEvtEventHandleT eventHandle,
  37                 const SaSizeT eventDataSize);
  38
  39 SaEvtCallbacksT callbacks = {
  40         0,
  41         event_callback
  42 };
  43
  44
  45 char channel[256] = "EVENT_TEST_CHANNEL";
  46 unsigned int subscription_id = 0xfedcba98;
  47 unsigned long long ret_time = 30000000000ULL; /* 30 seconds */
  48 char pubname[256] = "Test Pub Name";
  49
  50 #define _patt1 "Filter pattern 1"
  51 #define patt1 (SaUint8T *) _patt1
  52 #define patt1_size sizeof(_patt1)
  53
  54 #define _patt2 "Filter pattern 2"
  55 #define patt2 (SaUint8T *) _patt2
  56 #define patt2_size sizeof(_patt2)
  57
  58 #define _patt3 "Filter pattern 3"
  59 #define patt3 (SaUint8T *) _patt3
  60 #define patt3_size sizeof(_patt3)
  61
  62 #define _patt4 "Filter pattern 4"
  63 #define patt4 (SaUint8T *) _patt4
  64 #define patt4_size sizeof(_patt4)
  65
  66
  67 SaEvtEventFilterT filters[] = {
  68         {SA_EVT_PREFIX_FILTER, {patt1_size, patt1_size, patt1}},
  69         {SA_EVT_SUFFIX_FILTER, {patt2_size, patt2_size, patt2}},
  70         {SA_EVT_EXACT_FILTER, {patt3_size, patt3_size, patt3}},
  71         {SA_EVT_PASS_ALL_FILTER, {patt4_size, patt4_size, patt4}}
  72 };
  73
  74 SaEvtEventFilterArrayT subscribe_filters = {
  75         sizeof(filters)/sizeof(SaEvtEventFilterT),
  76         filters,
  77 };
  78
  79
  80         SaUint8T pat0[100];
  81         SaUint8T pat1[100];
  82         SaUint8T pat2[100];
  83         SaUint8T pat3[100];
  84         SaUint8T pat4[100];
  85         SaEvtEventPatternT evt_patts[5] = {
  86                 {100, 100, pat0},
  87                 {100, 100, pat1},
  88                 {100, 100, pat2},
  89                 {100, 100, pat3},
  90                 {100, 100, pat4}};
  91         SaEvtEventPatternArrayT evt_pat_get_array = { 5, 0, evt_patts };
  92
  93 SaEvtEventPatternT patterns[] = {
  94         {patt1_size, patt1_size, patt1},
  95         {patt2_size, patt2_size, patt2},
  96         {patt3_size, patt3_size, patt3},
  97         {patt4_size, patt4_size, patt4}
  98 };
  99 SaNameT test_pub_name;
 100 #define TEST_PRIORITY 2
 101
 102 SaEvtEventPatternArrayT evt_pat_set_array = {
 103         sizeof(patterns)/sizeof(SaEvtEventPatternT),
 104         sizeof(patterns)/sizeof(SaEvtEventPatternT),
 105         patterns
 106 };
 107
 108 char user_data_file[256];
 109 char  user_data[65536];
 110 int user_data_size = 0;
 111
 112 uint64_t clust_time_now(void)
 113 {
 114         struct timeval tv;
 115         uint64_t time_now;
 116
 117         if (gettimeofday(&tv, 0)) {
 118                 return 0ULL;
 119         }
 120
 121         time_now = (uint64_t)(tv.tv_sec) * 1000000000ULL;
 122         time_now += (uint64_t)(tv.tv_usec) * 1000ULL;
 123
 124         return time_now;
 125 }
 126
 127 int
 128 test_pub()
 129 {
 130         SaEvtHandleT handle;
 131         SaEvtChannelHandleT channel_handle;
 132         SaEvtEventHandleT event_handle;
 133         SaEvtChannelOpenFlagsT flags;
 134         SaNameT channel_name;
 135         uint64_t test_retention;
 136         SaSelectionObjectT fd;
 137         int i;
 138
 139         SaEvtEventIdT event_id;
 140 #ifdef EVENT_SUBSCRIBE
 141         struct pollfd pfd;
 142         int nfd;
 143         int timeout = 1000;
 144 #endif
 145
 146
 147
 148         SaAisErrorT result;
 149
 150         flags = SA_EVT_CHANNEL_PUBLISHER |
 151 #ifdef EVENT_SUBSCRIBE
 152                 SA_EVT_CHANNEL_SUBSCRIBER |
 153 #endif
 154                 SA_EVT_CHANNEL_CREATE;
 155         strcpy((char *)channel_name.value, channel);
 156         channel_name.length = strlen(channel);
 157
 158
 159         do {
 160                 result = saEvtInitialize (&handle, &callbacks, &version);
 161         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 162         if (result != SA_AIS_OK) {
 163                 get_sa_error(result, result_buf, result_buf_len);
 164                 printf("Event Initialize result: %s\n", result_buf);
 165                 return(result);
 166         }
 167         do {
 168                 result = saEvtChannelOpen(handle, &channel_name, flags,
 169                                 SA_TIME_MAX, &channel_handle);
 170         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 171         if (result != SA_AIS_OK) {
 172                 get_sa_error(result, result_buf, result_buf_len);
 173                 printf("channel open result: %s\n", result_buf);
 174                 return(result);
 175         }
 176
 177         /*
 178          * Publish with pattens
 179          */
 180         printf("Publish\n");
 181
 182 #ifdef EVENT_SUBSCRIBE
 183         do {
 184                 result = saEvtEventSubscribe(channel_handle,
 185                         &subscribe_filters,
 186                         subscription_id);
 187         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 188
 189         if (result != SA_AIS_OK) {
 190                 get_sa_error(result, result_buf, result_buf_len);
 191                 printf("event subscribe result: %s\n", result_buf);
 192                 return(result);
 193         }
 194 #endif
 195         do {
 196                 result = saEvtEventAllocate(channel_handle, &event_handle);
 197         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 198         if (result != SA_AIS_OK) {
 199                 get_sa_error(result, result_buf, result_buf_len);
 200                 printf("event Allocate result: %s\n", result_buf);
 201                 return(result);
 202         }
 203
 204         strcpy((char *)test_pub_name.value, pubname);
 205         test_pub_name.length = strlen(pubname);
 206         test_retention = ret_time;
 207         do {
 208                 result = saEvtEventAttributesSet(event_handle,
 209                         &evt_pat_set_array,
 210                         TEST_PRIORITY,
 211                         test_retention,
 212                         &test_pub_name);
 213         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 214         if (result != SA_AIS_OK) {
 215                 get_sa_error(result, result_buf, result_buf_len);
 216                 printf("event set attr result(2): %s\n", result_buf);
 217                 return(result);
 218         }
 219
 220         for (i = 0; i < pub_count; i++) {
 221                 do {
 222                         result = saEvtEventPublish(event_handle, user_data,
 223                                         user_data_size, &event_id);
 224                 } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 225                 if (result != SA_AIS_OK) {
 226                         get_sa_error(result, result_buf, result_buf_len);
 227                         printf("event Publish result(2): %s\n", result_buf);
 228                         return(result);
 229                 }
 230                 printf("Published event ID: 0x%llx\n",
 231                                 (unsigned long long)event_id);
 232         }
 233
 234         /*
 235          * See if we got the event
 236          */
 237         do {
 238                 result = saEvtSelectionObjectGet(handle, &fd);
 239         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 240         if (result != SA_AIS_OK) {
 241                 get_sa_error(result, result_buf, result_buf_len);
 242                 printf("saEvtSelectionObject get %s\n", result_buf);
 243                 /* error */
 244                 return(result);
 245         }
 246 #ifdef EVENT_SUBSCRIBE
 247
 248         for (i = 0; i < pub_count; i++) {
 249                 pfd.fd = fd;
 250                 pfd.events = POLLIN;
 251                 nfd = poll(&pfd, 1, timeout);
 252                 if (nfd <= 0) {
 253                         printf("poll fds %d\n", nfd);
 254                         if (nfd < 0) {
 255                                 perror("poll error");
 256                         }
 257                         goto evt_free;
 258                 }
 259
 260                 printf("Got poll event\n");
 261                 do {
 262                         result = saEvtDispatch(handle, SA_DISPATCH_ONE);
 263                 } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 264                 if (result != SA_AIS_OK) {
 265                         get_sa_error(result, result_buf, result_buf_len);
 266                         printf("saEvtDispatch %s\n", result_buf);
 267                         return(result);
 268                 }
 269         }
 270 #endif
 271
 272
 273         /*
 274          * Test cleanup
 275          */
 276         do {
 277                 result = saEvtEventFree(event_handle);
 278         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 279         if (result != SA_AIS_OK) {
 280                 get_sa_error(result, result_buf, result_buf_len);
 281                 printf("event free result: %s\n", result_buf);
 282                 return(result);
 283         }
 284
 285         do {
 286                 result = saEvtChannelClose(channel_handle);
 287         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 288
 289         if (result != SA_AIS_OK) {
 290                 get_sa_error(result, result_buf, result_buf_len);
 291                 printf("channel close result: %s\n", result_buf);
 292                 return(result);
 293         }
 294         do {
 295                 result = saEvtFinalize(handle);
 296         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 297
 298         if (result != SA_AIS_OK) {
 299                 get_sa_error(result, result_buf, result_buf_len);
 300                 printf("Event Finalize result: %s\n", result_buf);
 301                 return(result);
 302         }
 303         printf("Done\n");
 304         return SA_AIS_OK;
 305
 306 }
 307
 308 void
 309 event_callback( SaEvtSubscriptionIdT subscription_id,
 310                 const SaEvtEventHandleT event_handle,
 311                 const SaSizeT event_data_size)
 312 {
 313         SaAisErrorT result;
 314         SaUint8T priority;
 315         SaTimeT retention_time;
 316         SaNameT publisher_name = {0, {0}};
 317         SaTimeT publish_time;
 318         SaEvtEventIdT event_id;
 319         int i;
 320
 321         printf("event_callback called\n");
 322         printf("sub ID: %x\n", subscription_id);
 323         printf("event_handle %llx\n",
 324                         (unsigned long long)event_handle);
 325         printf("event data size %llu\n", (unsigned long long)event_data_size);
 326
 327         evt_pat_get_array.patternsNumber = 4;
 328         do {
 329                 result = saEvtEventAttributesGet(event_handle,
 330                         &evt_pat_get_array,     /* patterns */
 331                         &priority,              /* priority */
 332                         &retention_time,        /* retention time */
 333                         &publisher_name,        /* publisher name */
 334                         &publish_time,          /* publish time */
 335                         &event_id               /* event_id */
 336                         );
 337         } while ((result == SA_AIS_ERR_TRY_AGAIN) && !sleep(TRY_WAIT));
 338         if (result != SA_AIS_OK) {
 339                 get_sa_error(result, result_buf, result_buf_len);
 340                 printf("event get attr result(2): %s\n", result_buf);
 341                 goto evt_free;
 342         }
 343         printf("pattern array count: %llu\n",
 344                 (unsigned long long)evt_pat_get_array.patternsNumber);
 345         for (i = 0; i < evt_pat_get_array.patternsNumber; i++) {
 346                 printf( "pattern %d =\"%s\"\n", i,
 347                                   evt_pat_get_array.patterns[i].pattern);
 348         }
 349
 350         printf("priority: 0x%x\n", priority);
 351         printf("retention: 0x%llx\n",
 352                         (unsigned long long)retention_time);
 353         printf("publisher name content: \"%s\"\n", publisher_name.value);
 354         printf("event id: 0x%llx\n",
 355                         (unsigned long long)event_id);
 356 evt_free:
 357         result = saEvtEventFree(event_handle);
 358         get_sa_error(result, result_buf, result_buf_len);
 359         printf("event free result: %s\n", result_buf);
 360 }
 361
 362
 363 static int err_wait_time = -1;
 364
 365 int main (int argc, char **argv)
 366 {
 367         static const char opts[] = "c:i:t:n:x:u:w:f:";
 368
 369         int ret;
 370         int option;
 371
 372         while (1) {
 373                 option = getopt(argc, argv, opts);
 374                 if (option == -1)
 375                         break;
 376
 377                 switch (option) {
 378                 case 'u': {
 379                         int fd;
 380                         int sz;
 381
 382                         strcpy(user_data_file, optarg);
 383                         fd = open(user_data_file, O_RDONLY);
 384                         if (fd < 0) {
 385                                 printf("Can't open user data file %s\n",
 386                                                 user_data_file);
 387                                 exit(1);
 388                         }
 389                         sz = read(fd, user_data, 65536);
 390                         if (sz < 0) {
 391                                 perror("subscription\n");
 392                                 exit(1);
 393                         }
 394                         close(fd);
 395                         user_data_size = sz;
 396                         break;
 397                 }
 398
 399                 case 'c':
 400                         strcpy(channel, optarg);
 401                         break;
 402                 case 'n':
 403                         strcpy(pubname, optarg);
 404                         break;
 405                 case 'f':
 406                         err_wait_time =
 407                                 (unsigned int)strtoul(optarg, NULL, 0);
 408                         break;
 409                 case 'i':
 410                         subscription_id =
 411                                 (unsigned int)strtoul(optarg, NULL, 0);
 412                         break;
 413                 case 'w':
 414                         wait_time =
 415                                 (unsigned int)strtoul(optarg, NULL, 0);
 416                         break;
 417                 case 't':
 418                         ret_time = strtoull(optarg, NULL, 0);
 419                         ret_time *= 1000000000;
 420                         break;
 421                 case 'x':
 422                         pub_count = strtoul(optarg, NULL, 0);
 423                         break;
 424                 default:
 425                         printf("invalid arg: \"%s\"\n", optarg);
 426                         return 1;
 427                 }
 428         }
 429         do {
 430                 ret = test_pub();
 431                 if (ret != SA_AIS_OK) {
 432                         if (err_wait_time < 0) {
 433                                 exit(ret);
 434                         } else {
 435                                 sleep(err_wait_time);
 436                         }
 437                 } else if (wait_time < 0) {
 438                         break;
 439                 } else {
 440                         sleep(wait_time);
 441                 }
 442         } while(1);
 443         return 0;
 444 }