src/plugins/decision-proto/message.c

   1 #include "message.h"
   2
   3 static int append_one_row(mrp_msg_t *msg, uint16_t tag, mqi_column_def_t *col,
   4                           int ncolumn, mrp_pep_value_t *data);
   5
   6 mrp_msg_t *create_register_message(mrp_pep_t *pep)
   7 {
   8     mrp_msg_t        *msg;
   9     mrp_pep_table_t  *t;
  10     mqi_column_def_t *c;
  11     uint16_t          ncolumn, type;
  12     int               i, j;
  13
  14     ncolumn = 0;
  15     for (i = 0; i < pep->nowned; i++)
  16         ncolumn += pep->owned[i].ncolumn;
  17     for (i = 0; i < pep->nwatched; i++)
  18         ncolumn += pep->watched[i].ncolumn;
  19
  20     msg = mrp_msg_create(MRP_PEPMSG_UINT16(MSGTYPE, MRP_PEPMSG_REGISTER),
  21                          MRP_PEPMSG_UINT32(MSGSEQ , 0),
  22                          MRP_PEPMSG_STRING(NAME   , pep->name),
  23                          MRP_PEPMSG_UINT16(NTABLE , pep->nowned),
  24                          MRP_PEPMSG_UINT16(NWATCH , pep->nwatched),
  25                          MRP_PEPMSG_UINT16(NCOLDEF, ncolumn),
  26                          MRP_MSG_END);
  27
  28     for (i = 0, t = pep->owned; i < pep->nowned; i++, t++) {
  29         mrp_msg_append(msg, MRP_PEPMSG_STRING(TBLNAME, t->name));
  30         mrp_msg_append(msg, MRP_PEPMSG_UINT16(NCOLUMN, t->ncolumn));
  31         mrp_msg_append(msg, MRP_PEPMSG_SINT16(TBLIDX , t->idx_col));
  32         for (j = 0, c = t->columns; j < t->ncolumn; j++, c++) {
  33             if (c->type == mqi_varchar)
  34                 type = mqi_blob + c->length;
  35             else
  36                 type = c->type;
  37
  38             mrp_msg_append(msg, MRP_PEPMSG_STRING(COLNAME, c->name));
  39             mrp_msg_append(msg, MRP_PEPMSG_UINT16(COLTYPE, type));
  40         }
  41     }
  42
  43     for (i = 0, t = pep->watched; i < pep->nwatched; i++, t++) {
  44         mrp_msg_append(msg, MRP_PEPMSG_STRING(TBLNAME, t->name));
  45         mrp_msg_append(msg, MRP_PEPMSG_UINT16(NCOLUMN, t->ncolumn));
  46         for (j = 0, c = t->columns; j < t->ncolumn; j++, c++) {
  47             if (c->type == mqi_varchar)
  48                 type = mqi_blob + c->length;
  49             else
  50                 type = c->type;
  51
  52             mrp_msg_append(msg, MRP_PEPMSG_STRING(COLNAME, c->name));
  53             mrp_msg_append(msg, MRP_PEPMSG_UINT16(COLTYPE, type));
  54         }
  55     }
  56
  57
  58     return msg;
  59 }
  60
  61
  62 int decode_register_message(mrp_msg_t *msg, mrp_pep_table_t *owned, int nowned,
  63                             mrp_pep_table_t *watched, int nwatched,
  64                             mqi_column_def_t *columns, int ncolumn)
  65 {
  66     mrp_pep_table_t  *t;
  67     mqi_column_def_t *c;
  68     void             *it;
  69     char             *name;
  70     uint16_t          ntbl, nwch, ncol, type, idx_col;
  71     int               i, j, n;
  72
  73     it = NULL;
  74
  75     if (!mrp_msg_iterate_get(msg, &it,
  76                              MRP_PEPMSG_UINT16(NTABLE , &ntbl),
  77                              MRP_PEPMSG_UINT16(NWATCH , &nwch),
  78                              MRP_PEPMSG_UINT16(NCOLDEF, &ncol),
  79                              MRP_MSG_END))
  80         return FALSE;
  81
  82     if (ntbl > nowned || nwch > nwatched || ncol > ncolumn)
  83         return FALSE;
  84
  85     n = 0;
  86     c = columns;
  87     for (i = 0, t = owned; i < nowned; i++, t++) {
  88         if (mrp_msg_iterate_get(msg, &it,
  89                                 MRP_PEPMSG_STRING(TBLNAME, &name),
  90                                 MRP_PEPMSG_UINT16(NCOLUMN, &ncol),
  91                                 MRP_PEPMSG_SINT16(TBLIDX , &idx_col),
  92                                 MRP_MSG_END)) {
  93             t->name    = name;
  94             t->columns = c;
  95             t->ncolumn = ncol;
  96             t->idx_col = idx_col;
  97         }
  98         else
  99             return FALSE;
 100
 101         for (j = 0; j < t->ncolumn; j++, c++, n++) {
 102             if (n >= ncolumn)
 103                 return FALSE;
 104
 105             if (mrp_msg_iterate_get(msg, &it,
 106                                     MRP_PEPMSG_STRING(COLNAME, &name),
 107                                     MRP_PEPMSG_UINT16(COLTYPE, &type),
 108                                     MRP_MSG_END)) {
 109                 c->name = name;
 110
 111                 if (type > mqi_blob) {
 112                     c->type   = mqi_varchar;
 113                     c->length = type - mqi_blob;
 114                 }
 115                 else
 116                     c->type = type;
 117             }
 118         }
 119     }
 120
 121     for (i = 0, t = watched; i < nwatched; i++, t++) {
 122         if (mrp_msg_iterate_get(msg, &it,
 123                                 MRP_PEPMSG_STRING(TBLNAME, &name),
 124                                 MRP_PEPMSG_UINT16(NCOLUMN, &ncol),
 125                                 MRP_MSG_END)) {
 126             t->name    = name;
 127             t->columns = c;
 128             t->ncolumn = ncol;
 129             t->idx_col = -1;
 130         }
 131         else
 132             return FALSE;
 133
 134         for (j = 0; j < t->ncolumn; j++, c++, n++) {
 135             if (n >= ncolumn)
 136                 return FALSE;
 137             if (mrp_msg_iterate_get(msg, &it,
 138                                     MRP_PEPMSG_STRING(COLNAME, &name),
 139                                     MRP_PEPMSG_UINT16(COLTYPE, &type),
 140                                     MRP_MSG_END)) {
 141                 c->name = name;
 142
 143                 if (type > mqi_blob) {
 144                     c->type   = mqi_varchar;
 145                     c->length = type - mqi_blob;
 146                 }
 147                 else {
 148                     c->type   = type;
 149                     c->length = 0;
 150                 }
 151
 152                 c->flags = 0;
 153             }
 154         }
 155     }
 156
 157     return TRUE;
 158 }
 159
 160
 161 mrp_msg_t *create_ack_message(uint32_t seq)
 162 {
 163     return mrp_msg_create(MRP_PEPMSG_UINT16(MSGTYPE, MRP_PEPMSG_ACK),
 164                           MRP_PEPMSG_UINT32(MSGSEQ , seq),
 165                           MRP_MSG_END);
 166 }
 167
 168
 169 mrp_msg_t *create_nak_message(uint32_t seq, int error, const char *errmsg)
 170 {
 171     return mrp_msg_create(MRP_PEPMSG_UINT16(MSGTYPE, MRP_PEPMSG_NAK),
 172                           MRP_PEPMSG_UINT32(MSGSEQ , seq),
 173                           MRP_PEPMSG_SINT32(ERRCODE, error),
 174                           MRP_PEPMSG_STRING(ERRMSG , errmsg),
 175                           MRP_MSG_END);
 176 }
 177
 178
 179 mrp_msg_t *create_notify_message(void)
 180 {
 181     return mrp_msg_create(MRP_PEPMSG_UINT16(MSGTYPE, MRP_PEPMSG_NOTIFY),
 182                           MRP_PEPMSG_UINT32(MSGSEQ , 0),
 183                           MRP_PEPMSG_UINT16(NCHANGE, 0),
 184                           MRP_PEPMSG_UINT16(NTOTAL , 0),
 185                           MRP_MSG_END);
 186 }
 187
 188
 189 int update_notify_message(mrp_msg_t *msg, int id, mqi_column_def_t *columns,
 190                           int ncolumn, mrp_pep_value_t *data, int nrow)
 191 {
 192     mrp_pep_value_t *v;
 193     uint16_t         tid, nr;
 194     int              i;
 195
 196     nr  = nrow;
 197     tid = id;
 198
 199     if (!mrp_msg_append(msg, MRP_PEPMSG_UINT16(TBLID, tid)) ||
 200         !mrp_msg_append(msg, MRP_PEPMSG_UINT16(NROW , nr )))
 201         return FALSE;
 202
 203     for (i = 0, v = data; i < nrow; i++, v += ncolumn) {
 204         if (!append_one_row(msg, MRP_PEPTAG_DATA, columns, ncolumn, v))
 205             return FALSE;
 206     }
 207
 208     return TRUE;
 209 }
 210
 211
 212 int decode_notify_message(mrp_msg_t *msg, void **it, mrp_pep_data_t *data)
 213 {
 214     int               i, j;
 215     mrp_pep_value_t  *v;
 216     mqi_column_def_t *d;
 217
 218     v = data->columns;
 219     d = data->coldefs;
 220
 221     for (i = 0; i < data->nrow; i++) {
 222         for (j = 0; j < data->ncolumn; j++) {
 223             switch (d[j].type) {
 224             case mqi_varchar:
 225                 if (!mrp_msg_iterate_get(msg, it,
 226                                          MRP_PEPMSG_STRING(DATA, &v->str),
 227                                          MRP_MSG_END))
 228                     return FALSE;
 229                 break;
 230
 231             case mqi_integer:
 232                 if (!mrp_msg_iterate_get(msg, it,
 233                                          MRP_PEPMSG_SINT32(DATA, &v->s32),
 234                                          MRP_MSG_END))
 235                     return FALSE;
 236                 break;
 237
 238             case mqi_unsignd:
 239                 if (!mrp_msg_iterate_get(msg, it,
 240                                          MRP_PEPMSG_UINT32(DATA, &v->u32),
 241                                          MRP_MSG_END))
 242                     return FALSE;
 243                 break;
 244
 245             case mqi_floating:
 246                 if (!mrp_msg_iterate_get(msg, it,
 247                                          MRP_PEPMSG_DOUBLE(DATA, &v->dbl),
 248                                          MRP_MSG_END))
 249                     return FALSE;
 250                 break;
 251
 252             default:
 253                 return FALSE;
 254             }
 255
 256             v++;
 257         }
 258     }
 259
 260     return TRUE;
 261 }
 262
 263
 264 mrp_msg_t *create_set_message(uint32_t seq, mrp_pep_data_t *data, int ndata)
 265 {
 266     mrp_msg_t        *msg;
 267     mrp_pep_value_t  *vals;
 268     mqi_column_def_t *defs;
 269     uint16_t          utable, utotal, tid, nval, nrow;
 270     int               i, j;
 271
 272     utable = ndata;
 273     utotal = 0;
 274
 275     msg = mrp_msg_create(MRP_PEPMSG_UINT16(MSGTYPE, MRP_PEPMSG_SET),
 276                          MRP_PEPMSG_UINT32(MSGSEQ , seq),
 277                          MRP_PEPMSG_UINT16(NCHANGE, utable),
 278                          MRP_PEPMSG_UINT16(NTOTAL , 0),
 279                          MRP_MSG_END);
 280
 281     if (msg != NULL) {
 282         for (i = 0; i < ndata; i++) {
 283             tid  = data[i].id;
 284             vals = data[i].columns;
 285             defs = data[i].coldefs;
 286             nval = data[i].ncolumn;
 287             nrow = data[i].nrow;
 288
 289             if (!mrp_msg_append(msg, MRP_PEPMSG_UINT16(TBLID, tid)) ||
 290                 !mrp_msg_append(msg, MRP_PEPMSG_UINT16(NROW , nrow)))
 291                 goto fail;
 292
 293             for (j = 0; j < nrow; j++) {
 294                 if (!append_one_row(msg, MRP_PEPTAG_DATA, defs, nval, vals))
 295                     goto fail;
 296                 vals   += nval;
 297                 utotal += nval;
 298             }
 299         }
 300
 301         mrp_msg_set(msg, MRP_PEPMSG_UINT16(NTOTAL, utotal));
 302
 303         return msg;
 304     }
 305
 306  fail:
 307     mrp_msg_unref(msg);
 308     return NULL;
 309 }
 310
 311
 312 int decode_set_message(mrp_msg_t *msg, void **it, mrp_pep_data_t *data)
 313 {
 314     int               i, j;
 315     mrp_pep_value_t  *v;
 316     mqi_column_def_t *d;
 317
 318     v = data->columns;
 319     d = data->coldefs;
 320
 321     for (i = 0; i < data->nrow; i++) {
 322         for (j = 0; j < data->ncolumn; j++) {
 323             switch (d[j].type) {
 324             case mqi_varchar:
 325                 if (!mrp_msg_iterate_get(msg, it,
 326                                          MRP_PEPMSG_STRING(DATA, &v->str),
 327                                          MRP_MSG_END))
 328                     return FALSE;
 329                 break;
 330
 331             case mqi_integer:
 332                 if (!mrp_msg_iterate_get(msg, it,
 333                                          MRP_PEPMSG_SINT32(DATA, &v->s32),
 334                                          MRP_MSG_END))
 335                     return FALSE;
 336                 break;
 337
 338             case mqi_unsignd:
 339                 if (!mrp_msg_iterate_get(msg, it,
 340                                          MRP_PEPMSG_UINT32(DATA, &v->u32),
 341                                          MRP_MSG_END))
 342                     return FALSE;
 343                 break;
 344
 345             case mqi_floating:
 346                 if (!mrp_msg_iterate_get(msg, it,
 347                                          MRP_PEPMSG_DOUBLE(DATA, &v->dbl),
 348                                          MRP_MSG_END))
 349                     return FALSE;
 350                 break;
 351
 352             default:
 353                 return FALSE;
 354             }
 355
 356             v++;
 357         }
 358     }
 359
 360     return TRUE;
 361 }
 362
 363
 364 static int append_one_row(mrp_msg_t *msg, uint16_t tag, mqi_column_def_t *col,
 365                           int ncolumn, mrp_pep_value_t *data)
 366 {
 367 #define HANDLE_TYPE(dbtype, type, member)                                 \
 368     case mqi_##dbtype:                                                    \
 369         if (!mrp_msg_append(msg, MRP_MSG_TAG_##type(tag, data->member)))  \
 370             return FALSE;                                                 \
 371         break
 372
 373     int i;
 374
 375     for (i = 0; i < ncolumn; i++, data++, col++) {
 376         switch (col->type) {
 377             HANDLE_TYPE(integer , SINT32, s32);
 378             HANDLE_TYPE(unsignd , UINT32, u32);
 379             HANDLE_TYPE(floating, DOUBLE, dbl);
 380             HANDLE_TYPE(string  , STRING, str);
 381         case mqi_blob:
 382         default:
 383             return FALSE;
 384         }
 385     }
 386
 387     return TRUE;
 388
 389 #undef HANDLE_TYPE
 390 }