source/tpadmsv/queue.c

0001 /**
0002  * @brief Queue data grabber
0003  *
0004  * @file queue.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 <stdio.h>
0035 #include <stdlib.h>
0036 #include <string.h>
0037 #include <errno.h>
0038 #include <regex.h>
0039 #include <utlist.h>
0040 #include <unistd.h>
0041 #include <signal.h>
0042
0043 #include <ndebug.h>
0044 #include <atmi.h>
0045 #include <atmi_int.h>
0046 #include <typed_buf.h>
0047 #include <ndrstandard.h>
0048 #include <ubf.h>
0049 #include <Exfields.h>
0050 #include <Excompat.h>
0051 #include <ubfutil.h>
0052 #include <sys_unix.h>
0053
0054 #include "tpadmsv.h"
0055 #include "expr.h"
0056 /*---------------------------Externs------------------------------------*/
0057 /*---------------------------Macros-------------------------------------*/
0058 /*---------------------------Enums--------------------------------------*/
0059 /*---------------------------Typedefs-----------------------------------*/
0060
0061 /**
0062  * Image of the queue
0063  * + OID of rqaddr (q)
0064  */
0065 typedef struct
0066 {
0067     char lmid[MAXTIDENT+1];     /**< node id(last 30 chars)                 */
0068     char rqaddr[NDRX_MAX_Q_SIZE+1];/**< queue name (last 30 chars)          */
0069     char state[3+1];            /**< const: ACT - active machine            */
0070     long nqueued;               /**< Number of messages queued              */
0071     long rqid;                  /**< Queue ID (for System V only)           */
0072 } ndrx_adm_queue_t;
0073
0074
0075 /**
0076  * Client class infos mapping table
0077  */
0078 expublic ndrx_adm_elmap_t ndrx_G_queue_map[] =
0079 {
0080     /* Driving of the Preparing: */
0081      {TA_LMID,                  TPADM_EL(ndrx_adm_queue_t, lmid)}
0082     ,{TA_RQADDR,                TPADM_EL(ndrx_adm_queue_t, rqaddr)}
0083     ,{TA_STATE,                 TPADM_EL(ndrx_adm_queue_t, state)}
0084     ,{TA_NQUEUED,               TPADM_EL(ndrx_adm_queue_t, nqueued)}
0085     /* TA_RQID for System V mapping...
0086      * Resolve from ndrx_svqshm_get(char *qstr, mode_t mode, int oflag)
0087      */
0088     ,{TA_RQID,                  TPADM_EL(ndrx_adm_queue_t, rqid)}
0089     ,{BBADFLDID}
0090 };
0091
0092 /*---------------------------Globals------------------------------------*/
0093 /*---------------------------Statics------------------------------------*/
0094 /*---------------------------Prototypes---------------------------------*/
0095
0096 /**
0097  * Read queue data
0098  * @param clazz class name
0099  * @param cursnew this is new cursor domain model
0100  * @param flags not used
0101  */
0102 expublic int ndrx_adm_queue_get(char *clazz, ndrx_adm_cursors_t *cursnew, long flags)
0103 {
0104     int ret = EXSUCCEED;
0105     string_list_t* qlist = NULL;
0106     string_list_t* elt = NULL;
0107     ndrx_adm_queue_t q;
0108     int idx = 0;
0109     struct mq_attr att;
0110
0111     cursnew->map = ndrx_G_queue_map;
0112
0113     ndrx_growlist_init(&cursnew->list, 100, sizeof(ndrx_adm_queue_t));
0114
0115     qlist = ndrx_sys_mqueue_list_make(G_atmi_env.qpath, &ret);
0116
0117     if (EXSUCCEED!=ret)
0118     {
0119         NDRX_LOG(log_error, "posix queue listing failed!");
0120         EXFAIL_OUT(ret);
0121     }
0122
0123     /* get the usage states / check queues... */
0124     LL_FOREACH(qlist,elt)
0125     {
0126         /* if not print all, then skip this queue */
0127         if (0!=strncmp(elt->qname,
0128                 G_atmi_env.qprefix_match, G_atmi_env.qprefix_match_len))
0129         {
0130             continue;
0131         }
0132
0133         memset(&q, 0, sizeof(q));
0134
0135         NDRX_STRCPY_SAFE(q.rqaddr, elt->qname);
0136
0137 #if defined(EX_USE_SYSVQ) || defined(EX_USE_SVAPOLL)
0138         q.rqid = ndrx_svqshm_get(q.rqaddr, 0, 0);
0139 #endif
0140
0141         snprintf(q.lmid, sizeof(q.lmid), "%ld", tpgetnodeid());
0142         NDRX_STRCPY_SAFE(q.state, "ACT");
0143
0144         /* get stats... */
0145         if (EXSUCCEED!=ndrx_get_q_attr(elt->qname, &att))
0146         {
0147             /* skip this one... */
0148             continue;
0149         }
0150
0151         q.nqueued = att.mq_curmsgs;
0152
0153         if (EXSUCCEED!=ndrx_growlist_add(&cursnew->list, (void *)&q, idx))
0154         {
0155             NDRX_LOG(log_error, "Growlist on queue failed - out of memory?");
0156             EXFAIL_OUT(ret);
0157         }
0158
0159         idx++;
0160     }
0161
0162 out:
0163     if (NULL!=qlist)
0164     {
0165         ndrx_string_list_free(qlist);
0166     }
0167
0168     if (EXSUCCEED!=ret)
0169     {
0170         ndrx_growlist_free(&cursnew->list);
0171     }
0172
0173     return ret;
0174 }
0175
0176 /* vim: set ts=4 sw=4 et smartindent: */