atmitest/test039_tpbroadcast/atmiclt39.c

0001 /**
0002  * @brief tpbroadcast tests
0003  *
0004  * @file atmiclt39.c
0005  */
0006 /* -----------------------------------------------------------------------------
0007  * Enduro/X Middleware Platform for Distributed Transaction Processing
0008  * Copyright (C) 2009-2016, ATR Baltic, Ltd. All Rights Reserved.
0009  * Copyright (C) 2017-2023, Mavimax, Ltd. All Rights Reserved.
0010  * This software is released under one of the following licenses:
0011  * AGPL (with Java and Go exceptions) or Mavimax's license for commercial use.
0012  * See LICENSE file for full text.
0013  * -----------------------------------------------------------------------------
0014  * AGPL license:
0015  *
0016  * This program is free software; you can redistribute it and/or modify it under
0017  * the terms of the GNU Affero General Public License, version 3 as published
0018  * by the Free Software Foundation;
0019  *
0020  * This program is distributed in the hope that it will be useful, but WITHOUT ANY
0021  * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
0022  * PARTICULAR PURPOSE. See the GNU Affero General Public License, version 3
0023  * for more details.
0024  *
0025  * You should have received a copy of the GNU Affero General Public License along
0026  * with this program; if not, write to the Free Software Foundation, Inc.,
0027  * 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
0028  *
0029  * -----------------------------------------------------------------------------
0030  * A commercial use license is available from Mavimax, Ltd
0031  * contact@mavimax.com
0032  * -----------------------------------------------------------------------------
0033  */
0034 #include <string.h>
0035 #include <stdio.h>
0036 #include <stdlib.h>
0037 #include <memory.h>
0038 #include <signal.h>
0039
0040 #include <atmi.h>
0041 #include <ubf.h>
0042 #include <ndebug.h>
0043 #include <test.fd.h>
0044 #include <ndrstandard.h>
0045 #include <ubfutil.h>
0046 /*---------------------------Externs------------------------------------*/
0047 /*---------------------------Macros-------------------------------------*/
0048 #define MAX_CALLS       1000
0049 /*---------------------------Enums--------------------------------------*/
0050 /*---------------------------Typedefs-----------------------------------*/
0051 /*---------------------------Globals------------------------------------*/
0052 /*---------------------------Statics------------------------------------*/
0053 exprivate int M_shutdown = EXFALSE;
0054 /*---------------------------Prototypes---------------------------------*/
0055
0056 int test_tpchkunsol_ret_num(void);
0057
0058 /**
0059  * Set notification handler
0060  * @return
0061  */
0062 void notification_callback (char *data, long len, long flags)
0063 {
0064     UBFH *p_ub = (UBFH *)data;
0065     NDRX_LOG(log_info, "Got broadcast...");
0066     /* Dump UBF buffer... */
0067     ndrx_debug_dump_UBF(log_error, "notification_callback", p_ub);
0068 }
0069
0070 /**
0071  * Do the broadcasting
0072  */
0073 int run_broadcast(void)
0074 {
0075     int ret = EXSUCCEED;
0076     UBFH *p_ub = NULL;
0077     char tmp[32];
0078     int i;
0079
0080     if (EXSUCCEED!=tpinit(NULL))
0081     {
0082         NDRX_LOG(log_error, "TESTERROR: Failed to init!!!!");
0083         EXFAIL_OUT(ret);
0084     }
0085
0086     if (NULL!=tpsetunsol(notification_callback))
0087     {
0088         NDRX_LOG(log_error, "TESTERRORR: Previous handler must be NULL!");
0089         EXFAIL_OUT(ret);
0090     }
0091
0092     /* Allocate some buffer */
0093     if (NULL==(p_ub = (UBFH *)tpalloc("UBF", NULL, 1024)))
0094     {
0095         NDRX_LOG(log_error, "TESTERROR: Failed to allocate test buffer!");
0096         EXFAIL_OUT(ret);
0097     }
0098
0099     /**
0100      * Send simple broadcast to all nodes and clients...
0101      */
0102     NDRX_LOG(log_info, ">>> Send simple broadcast to all nodes and clients...");
0103     for (i=0; i<MAX_CALLS; i++)
0104     {
0105         /* Let servers to provide back replies... */
0106         /* Have a fixed buffer len, so that we do not need to realloc... */
0107
0108         snprintf(tmp, sizeof(tmp), "AA%02ld%08d", tpgetnodeid(), i);
0109
0110         if (EXSUCCEED!=Bchg(p_ub, T_STRING_FLD, 0, tmp, 0L))
0111         {
0112             NDRX_LOG(log_error, "TESTERROR: Failed to set T_STRING_FLD to [%s]: %s",
0113                     tmp, Bstrerror(Berror));
0114             EXFAIL_OUT(ret);
0115         }
0116
0117         if (i % 100)
0118         {
0119             usleep(500);
0120         }
0121
0122         /* generic broadcast applied to all machines... */
0123         /* seems like we are doing broadcasts to tpbridge threads too..
0124          * thus will send only to atmi matched clients... */
0125         if (EXSUCCEED!=tpbroadcast(NULL, NULL, "atmi", (char *)p_ub, 0L, TPREGEXMATCH))
0126         {
0127             NDRX_LOG(log_error, "TESTERRROR: Failed to broadcast: %s",
0128                     tpstrerror(tperrno));
0129             EXFAIL_OUT(ret);
0130         }
0131
0132         if (EXFAIL==tpchkunsol())
0133         {
0134             NDRX_LOG(log_error, "TESTERROR: tpchkunsol() failed!");
0135             EXFAIL_OUT(ret);
0136         }
0137     }
0138
0139     /* Send only to B like services */
0140     NDRX_LOG(log_info, ">>> Send only to B like services");
0141     for (i=0; i<MAX_CALLS; i++)
0142     {
0143         /* Let servers to provide back replies... */
0144         /* Have a fixed buffer len, so that we do not need to realloc... */
0145
0146         snprintf(tmp, sizeof(tmp), "BB%02ld%08d", tpgetnodeid(), i);
0147
0148         if (EXSUCCEED!=Bchg(p_ub, T_STRING_FLD, 0, tmp, 0L))
0149         {
0150             NDRX_LOG(log_error, "TESTERROR: Failed to set T_STRING_FLD to [%s]: %s",
0151                     tmp, Bstrerror(Berror));
0152             EXFAIL_OUT(ret);
0153         }
0154
0155         if (i % 100)
0156         {
0157             usleep(500);
0158         }
0159
0160         /* generic broadcast applied to all machines... */
0161         if (EXSUCCEED!=tpbroadcast(NULL, NULL, "^.*B.*", (char *)p_ub, 0L, TPREGEXMATCH))
0162         {
0163             NDRX_LOG(log_error, "TESTERRROR: Failed to broadcast: %s",
0164                     tpstrerror(tperrno));
0165             EXFAIL_OUT(ret);
0166         }
0167
0168         if (EXFAIL==tpchkunsol())
0169         {
0170             NDRX_LOG(log_error, "TESTERROR: tpchkunsol() failed!");
0171             EXFAIL_OUT(ret);
0172         }
0173     }
0174
0175     /* Send only to 2nd node */
0176     NDRX_LOG(log_info, ">>> Send only to 2nd node");
0177     for (i=0; i<MAX_CALLS; i++)
0178     {
0179         /* Let servers to provide back replies... */
0180         /* Have a fixed buffer len, so that we do not need to realloc... */
0181
0182         snprintf(tmp, sizeof(tmp), "CC%02ld%08d", tpgetnodeid(), i);
0183
0184         if (EXSUCCEED!=Bchg(p_ub, T_STRING_FLD, 0, tmp, 0L))
0185         {
0186             NDRX_LOG(log_error, "TESTERROR: Failed to set T_STRING_FLD to [%s]: %s",
0187                     tmp, Bstrerror(Berror));
0188             EXFAIL_OUT(ret);
0189         }
0190
0191         if (i % 100)
0192         {
0193             usleep(500);
0194         }
0195
0196         /* generic broadcast applied to all machines... */
0197         if (EXSUCCEED!=tpbroadcast("2", NULL, "atmi", (char *)p_ub, 0L, TPREGEXMATCH))
0198         {
0199             NDRX_LOG(log_error, "TESTERRROR: Failed to broadcast: %s",
0200                     tpstrerror(tperrno));
0201             EXFAIL_OUT(ret);
0202         }
0203
0204         if (EXFAIL==tpchkunsol())
0205         {
0206             NDRX_LOG(log_error, "TESTERROR: tpchkunsol() failed!");
0207             EXFAIL_OUT(ret);
0208         }
0209
0210     }
0211
0212     /* Send to single client (matching the name) */
0213     NDRX_LOG(log_info, ">>> Send to single client (matching the name)");
0214     for (i=0; i<MAX_CALLS; i++)
0215     {
0216         /* Let servers to provide back replies... */
0217         /* Have a fixed buffer len, so that we do not need to realloc... */
0218
0219         snprintf(tmp, sizeof(tmp), "DD%02ld%08d", tpgetnodeid(), i);
0220
0221         if (EXSUCCEED!=Bchg(p_ub, T_STRING_FLD, 0, tmp, 0L))
0222         {
0223             NDRX_LOG(log_error, "TESTERROR: Failed to set T_STRING_FLD to [%s]: %s",
0224                     tmp, Bstrerror(Berror));
0225             EXFAIL_OUT(ret);
0226         }
0227
0228         if (i % 100)
0229         {
0230             usleep(500);
0231         }
0232
0233         /* generic broadcast applied to all machines... */
0234         if (EXSUCCEED!=tpbroadcast(NULL, NULL, "atmicltC39", (char *)p_ub, 0L, 0L))
0235         {
0236             NDRX_LOG(log_error, "TESTERRROR: Failed to broadcast: %s",
0237                     tpstrerror(tperrno));
0238             EXFAIL_OUT(ret);
0239         }
0240
0241         if (EXFAIL==tpchkunsol())
0242         {
0243             NDRX_LOG(log_error, "TESTERROR: tpchkunsol() failed!");
0244             EXFAIL_OUT(ret);
0245         }
0246     }
0247
0248 out:
0249
0250     if (NULL!=p_ub)
0251     {
0252         tpfree((char *)p_ub);
0253     }
0254     return ret;
0255
0256 }
0257
0258 /**
0259  * Run listener
0260  */
0261 int bc_listen(void)
0262 {
0263     int ret = EXSUCCEED;
0264
0265     if (NULL!=tpsetunsol(notification_callback))
0266     {
0267         NDRX_LOG(log_error, "TESTERRORR: Previous handler must be NULL!");
0268         EXFAIL_OUT(ret);
0269     }
0270
0271     while (!M_shutdown)
0272     {
0273         int applied;
0274         while (EXFAIL!=(applied=tpchkunsol()))
0275         {
0276             if (applied > 0)
0277             {
0278                 NDRX_LOG(log_debug, "Applied: %d", applied);
0279             }
0280             /* Have some sleep */
0281             usleep(100000);
0282         }
0283
0284         if (EXFAIL==applied)
0285         {
0286             NDRX_LOG(log_error, "TESTERROR: failed to call tpchkunsol(): %s",
0287                     tpstrerror(tperrno));
0288             EXFAIL_OUT(ret);
0289         }
0290     }
0291
0292 out:
0293     return ret;
0294 }
0295 /**
0296  * Check the number of events received
0297  * Bug #269
0298  * @return EXSUCCEED/EXFAIL
0299  */
0300 int test_tpchkunsol_ret_num(void)
0301 {
0302     int ret = EXSUCCEED;
0303     UBFH *p_ub = NULL;
0304     char tmp[32];
0305     int i;
0306     int cnt;
0307
0308     if (NULL!=tpsetunsol(notification_callback))
0309     {
0310         NDRX_LOG(log_error, "TESTERRORR: Previous handler must be NULL!");
0311         EXFAIL_OUT(ret);
0312     }
0313
0314     /* Allocate some buffer */
0315     if (NULL==(p_ub = (UBFH *)tpalloc("UBF", NULL, 1024)))
0316     {
0317         NDRX_LOG(log_error, "TESTERROR: Failed to allocate test buffer!");
0318         EXFAIL_OUT(ret);
0319     }
0320 #define NUMCALLS    7
0321
0322     for (i=0; i<NUMCALLS; i++)
0323     {
0324         /* Let servers to provide back replies... */
0325         /* Have a fixed buffer len, so that we do not need to realloc... */
0326
0327         snprintf(tmp, sizeof(tmp), "AA%02ld%08d", tpgetnodeid(), i);
0328
0329         if (EXSUCCEED!=Bchg(p_ub, T_STRING_FLD, 0, tmp, 0L))
0330         {
0331             NDRX_LOG(log_error, "TESTERROR: Failed to set T_STRING_FLD to [%s]: %s",
0332                     tmp, Bstrerror(Berror));
0333             EXFAIL_OUT(ret);
0334         }
0335
0336         if (EXSUCCEED!=tpbroadcast(NULL, NULL, "atmicltA39", (char *)p_ub, 0L, TPREGEXMATCH))
0337         {
0338             NDRX_LOG(log_error, "TESTERRROR: Failed to broadcast: %s",
0339                     tpstrerror(tperrno));
0340             EXFAIL_OUT(ret);
0341         }
0342
0343     }
0344
0345     sleep(10);
0346
0347     if (NUMCALLS!=(cnt=tpchkunsol()))
0348     {
0349         NDRX_LOG(log_error, "TESTERROR: Expected numcalls: %d but got %d!!!",
0350                 NUMCALLS, cnt);
0351         EXFAIL_OUT(ret);
0352     }
0353
0354 out:
0355     return ret;
0356 }
0357
0358 void sighandler(int signum)
0359 {
0360     M_shutdown = EXTRUE;
0361 }
0362
0363 /**
0364  * Do the test call to the server
0365  */
0366 int main(int argc, char** argv)
0367 {
0368     int ret = EXSUCCEED;
0369     TPINIT init;
0370
0371     if (argc<2)
0372     {
0373         NDRX_LOG(log_error, "usage: %s <broadcast|listen|mutted|>");
0374         EXFAIL_OUT(ret);
0375     }
0376
0377     signal(SIGINT, sighandler);
0378     signal(SIGTERM, sighandler);
0379
0380     memset(&init, 0, sizeof(init));
0381
0382     if (0==strcmp(argv[1], "retnum"))
0383     {
0384         NDRX_LOG(log_error, "Running: broadcast");
0385         ret = test_tpchkunsol_ret_num();
0386     }
0387     else if (0==strcmp(argv[1], "broadcast"))
0388     {
0389         NDRX_LOG(log_error, "Running: broadcast");
0390         ret = run_broadcast();
0391     }
0392     else if (0==strcmp(argv[1], "listen"))
0393     {
0394         NDRX_LOG(log_error, "Running: listen");
0395         /* no flags.. */
0396         if (EXSUCCEED!=tpinit(&init))
0397         {
0398             NDRX_LOG(log_error, "TESTERROR: Failed to init!!!!");
0399             EXFAIL_OUT(ret);
0400         }
0401
0402         ret = bc_listen();
0403     }
0404     else if (0==strcmp(argv[1], "mutted"))
0405     {
0406         NDRX_LOG(log_error, "Running: mutted");
0407         init.flags|=TPU_IGN;
0408         if (EXSUCCEED!=tpinit(&init))
0409         {
0410             NDRX_LOG(log_error, "TESTERROR: Failed to init!!!!");
0411             EXFAIL_OUT(ret);
0412         }
0413
0414         ret = bc_listen();
0415     }
0416
0417 out:
0418
0419     if (EXSUCCEED!=ret)
0420     {
0421         NDRX_LOG(log_error, "TESTERROR: main() finishing with error %d!", ret);
0422     }
0423
0424     tpterm();
0425
0426     return ret;
0427
0428 }
0429
0430
0431 /* vim: set ts=4 sw=4 et smartindent: */