confd_client/confd_client.c

   1
   2 /*
   3  *
   4  *   Copyright 2016 RIFT.IO Inc
   5  *
   6  *   Licensed under the Apache License, Version 2.0 (the "License");
   7  *   you may not use this file except in compliance with the License.
   8  *   You may obtain a copy of the License at
   9  *
  10  *       http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *   Unless required by applicable law or agreed to in writing, software
  13  *   distributed under the License is distributed on an "AS IS" BASIS,
  14  *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  *   See the License for the specific language governing permissions and
  16  *   limitations under the License.
  17  *
  18  *
  19  */
  20
  21 #include <stdio.h>
  22 #include <stdlib.h>
  23 #include <poll.h>
  24 #include <unistd.h>
  25 #include <string.h>
  26
  27 #include <confd_lib.h>
  28 #include "confd_cdb.h"
  29 #include "confd_dp.h"
  30
  31 static struct confd_daemon_ctx *dctx;
  32 static int ctlsock;
  33 static int workersock;
  34
  35 typedef struct _foodata {
  36   char *name;
  37   struct _foodata *next;
  38 } foodata_t;
  39
  40 typedef struct _opdata {
  41   foodata_t *foo;
  42 } opdata_t;
  43
  44 opdata_t *g_opdata = NULL;
  45
  46 int process_confd_subscription(int subsock)
  47 {
  48   int confd_result, flags, length, *subscription_points, i, j, nvalues;
  49   enum cdb_sub_notification type;
  50   confd_tag_value_t *values;
  51
  52   confd_result = cdb_read_subscription_socket2(subsock,
  53                                                &type,
  54                                                &flags,
  55                                                &subscription_points,
  56                                                &length);
  57
  58   if (confd_result != CONFD_OK) {
  59     confd_fatal("Failed to read subscription data \n");
  60   }
  61
  62   switch (type) {
  63     case CDB_SUB_PREPARE:
  64       for (i = 0; i < length; i++) {
  65         printf("i = %d, point = %d\n", i, subscription_points[i]);
  66         if (cdb_get_modifications(subsock, subscription_points[i], flags, &values, &nvalues,
  67                               "/") == CONFD_OK) {
  68           for (j = 0; j < nvalues; j++) {
  69             printf("j = %d\n", j);
  70             confd_free_value(CONFD_GET_TAG_VALUE(&values[j]));
  71           }
  72         }
  73       }
  74       cdb_sync_subscription_socket(subsock, CDB_DONE_PRIORITY);
  75       fprintf(stdout, "CBD_SUB_PREPARE\n");
  76       break;
  77
  78     case CDB_SUB_COMMIT:
  79       cdb_sync_subscription_socket(subsock, CDB_DONE_PRIORITY);
  80       fprintf(stdout, "CDB_SUB_COMMIT\n");
  81       break;
  82
  83     case CDB_SUB_ABORT:
  84       fprintf(stdout, "CDB_SUB_ABORT\n");
  85       break;
  86
  87     default:
  88       confd_fatal("Invalid type %d in cdb_read_subscription_socket2\n", type);
  89   }
  90
  91   return 0;
  92 }
  93
  94 static int do_init_action(struct confd_user_info *uinfo)
  95 {
  96   int ret = CONFD_OK;
  97   // fprintf(stdout, "init_action called\n");
  98   confd_action_set_fd(uinfo, workersock);
  99   return ret;
 100 }
 101
 102 static int do_rw_action(struct confd_user_info *uinfo,
 103                         struct xml_tag *name,
 104                         confd_hkeypath_t *kp,
 105                         confd_tag_value_t *params,
 106                         int nparams)
 107 {
 108   // confd_tag_value_t reply[2];
 109   // int status;
 110   // char *ret_status;
 111   int i;
 112   char buf[BUFSIZ];
 113
 114   /* Just print the parameters and return */
 115
 116   //
 117   for (i = 0; i < nparams; i++) {
 118     confd_pp_value(buf, sizeof(buf), CONFD_GET_TAG_VALUE(&params[i]));
 119     printf("param %2d: %9u:%-9u, %s\n", i, CONFD_GET_TAG_NS(&params[i]),
 120            CONFD_GET_TAG_TAG(&params[i]), buf);
 121   }
 122
 123   i = 0;
 124   // CONFD_SET_TAG_INT32(&reply[i], NULL, 0); i++;
 125   // CONFD_SET_TAG_STR(&reply[i], NULL, "success"); i++;
 126   confd_action_reply_values(uinfo, NULL, i);
 127
 128   return CONFD_OK;
 129
 130 }
 131
 132 static int get_next(struct confd_trans_ctx *tctx,
 133                     confd_hkeypath_t *keypath,
 134                     long next)
 135 {
 136   opdata_t *opdata = tctx->t_opaque;
 137   foodata_t *curr;
 138   confd_value_t v[2];
 139
 140   if (next == -1) { /* first call */
 141     curr = opdata->foo;
 142   } else {
 143     curr = (foodata_t *)next;
 144   }
 145
 146   if (curr == NULL) {
 147     confd_data_reply_next_key(tctx, NULL, -1, -1);
 148     return CONFD_OK;
 149   }
 150
 151   CONFD_SET_STR(&v[0], curr->name);
 152   confd_data_reply_next_key(tctx, &v[0], 1, (long)curr->next);
 153   return CONFD_OK;
 154 }
 155
 156 static foodata_t *find_foo(confd_hkeypath_t *keypath, opdata_t *dp)
 157 {
 158   char *name = (char*)CONFD_GET_BUFPTR(&keypath->v[1][0]);
 159   foodata_t *foo = dp->foo;
 160   while (foo != NULL) {
 161     if (strcmp(foo->name, name) == 0) {
 162       return foo;
 163     }
 164     foo = foo->next;
 165   }
 166   return NULL;
 167 }
 168
 169 /* Keypath example */
 170 /* /arpentries/arpe{192.168.1.1 eth0}/hwaddr */
 171 /* 3 2 1 0 */
 172 static int get_elem(struct confd_trans_ctx *tctx,
 173                     confd_hkeypath_t *keypath)
 174 {
 175   confd_value_t v;
 176   foodata_t *foo = find_foo(keypath, tctx->t_opaque);
 177   if (foo == NULL) {
 178     confd_data_reply_not_found(tctx);
 179     return CONFD_OK;
 180   }
 181
 182   CONFD_SET_STR(&v, foo->name);
 183   confd_data_reply_value(tctx, &v);
 184
 185   return CONFD_OK;
 186 }
 187
 188 static foodata_t *create_dummy_foodata_list(int count)
 189 {
 190   foodata_t *head, *curr, *prev;
 191   int i;
 192   char buf[64];
 193
 194   head = prev = curr = NULL;
 195   for (i = 0; i < count; ++i) {
 196     curr = malloc(sizeof(foodata_t));
 197     memset(curr, 0, sizeof(foodata_t));
 198     snprintf(buf, 64, "foo%d", i);
 199     curr->name = strdup(buf);
 200     if (prev) {
 201       prev->next = curr;
 202     } else {
 203       head = curr;
 204     }
 205     prev = curr;
 206   }
 207
 208   return head;
 209 }
 210
 211 static void free_foodata_list(foodata_t *foo)
 212 {
 213   foodata_t *curr, *next;
 214   curr = foo;
 215   while (curr) {
 216     next = curr->next;
 217     if (curr->name) {
 218       free(curr->name);
 219     }
 220     free(curr);
 221     curr = next;
 222   }
 223 }
 224
 225 static void print_foodata_list(foodata_t *foo)
 226 {
 227   foodata_t *curr = foo;
 228   while (curr) {
 229     // fprintf(stdout, "%s\n", curr->name);
 230     curr = curr->next;
 231   }
 232 }
 233
 234 static int s_init(struct confd_trans_ctx *tctx)
 235 {
 236   opdata_t *opdata;
 237   if ((opdata = malloc(sizeof(opdata_t))) == NULL) {
 238     return CONFD_ERR;
 239   }
 240
 241   memset(opdata, 0, sizeof(opdata_t));
 242   opdata->foo = create_dummy_foodata_list(10);
 243   print_foodata_list(opdata->foo);
 244   tctx->t_opaque = opdata;
 245   confd_trans_set_fd(tctx, workersock);
 246   return CONFD_OK;
 247 }
 248
 249 static int s_finish(struct confd_trans_ctx *tctx)
 250 {
 251   opdata_t *opdata = tctx->t_opaque;
 252   if (opdata != NULL) {
 253     free_foodata_list(opdata->foo);
 254     free(opdata);
 255   }
 256
 257   return CONFD_OK;
 258 }
 259
 260 int main(int argc, char **argv)
 261 {
 262   struct sockaddr_in addr;
 263   int debuglevel = CONFD_TRACE;
 264   struct confd_trans_cbs trans;
 265   struct confd_data_cbs data;
 266   struct confd_action_cbs action;
 267   int i;
 268
 269   int subsock, datasock;
 270   int status;
 271   int spoint;
 272
 273   addr.sin_addr.s_addr = inet_addr("127.0.0.1");
 274   addr.sin_family = AF_INET;
 275   addr.sin_port = htons(CONFD_PORT);
 276
 277    /**
 278    * Setup CDB subscription socket
 279    */
 280   confd_init(argv[0], stderr, CONFD_DEBUG);
 281   if ((subsock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 282     confd_fatal("Failed to open subscription socket\n");
 283   }
 284
 285   printf("Subscription socket: %d\n", subsock);
 286
 287   for (i = 1; i < 10; ++i) {
 288     if (cdb_connect(subsock, CDB_SUBSCRIPTION_SOCKET,
 289                     (struct sockaddr*)&addr,
 290                     sizeof (struct sockaddr_in)) < 0) {
 291       sleep(2);
 292       fprintf(stdout, "Failed in confd_connect() {attempt: %d}\n", i);
 293     } else {
 294       fprintf(stdout, "confd_connect succeeded\n");
 295       break;
 296     }
 297   }
 298
 299   if ((status = cdb_subscribe2(subsock, CDB_SUB_RUNNING_TWOPHASE, 0, 0, &spoint, 0, "/"))
 300       != CONFD_OK) {
 301     fprintf(stderr, "Terminate: subscribe %d\n", status);
 302     exit(1);
 303   }
 304
 305   if (cdb_subscribe_done(subsock) != CONFD_OK) {
 306     confd_fatal("cdb_subscribe_done() failed");
 307   }
 308
 309   /**
 310    * Setup CBD data socket
 311    */
 312
 313   if ((datasock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 314     confd_fatal("Failed to open data socket\n");
 315   }
 316
 317   if (cdb_connect(datasock, CDB_DATA_SOCKET,
 318                   (struct sockaddr*)&addr,
 319                   sizeof (struct sockaddr_in)) < 0) {
 320     confd_fatal("Failed to confd_connect() to confd \n");
 321   }
 322
 323   memset(&trans, 0, sizeof (struct confd_trans_cbs));
 324   trans.init = s_init;
 325   trans.finish = s_finish;
 326
 327   memset(&data, 0, sizeof (struct confd_data_cbs));
 328   data.get_elem = get_elem;
 329   data.get_next = get_next;
 330   strcpy(data.callpoint, "base_show");
 331
 332   memset(&action, 0, sizeof (action));
 333   strcpy(action.actionpoint, "rw_action");
 334   action.init = do_init_action;
 335   action.action = do_rw_action;
 336
 337
 338   /* initialize confd library */
 339   confd_init("confd_client_op_data_daemon", stderr, debuglevel);
 340
 341
 342   for (i = 1; i < 10; ++i) {
 343     if (confd_load_schemas((struct sockaddr*)&addr,
 344                            sizeof(struct sockaddr_in)) != CONFD_OK) {
 345       fprintf(stdout, "Failed to load schemas from confd {attempt: %d}\n", i);
 346       sleep(2);
 347     } else {
 348       fprintf(stdout, "confd_load_schemas succeeded\n");
 349       break;
 350     }
 351   }
 352
 353   if ((dctx = confd_init_daemon("confd_client_op_data_daemon")) == NULL) {
 354     confd_fatal("Failed to initialize confdlib\n");
 355   }
 356
 357   /* Create the first control socket, all requests to */
 358   /* create new transactions arrive here */
 359   if ((ctlsock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 360     confd_fatal("Failed to open ctlsocket\n");
 361   }
 362
 363   if (confd_connect(dctx, ctlsock, CONTROL_SOCKET, (struct sockaddr*)&addr,
 364                     sizeof (struct sockaddr_in)) < 0) {
 365     confd_fatal("Failed to confd_connect() to confd \n");
 366   }
 367
 368   /* Also establish a workersocket, this is the most simple */
 369   /* case where we have just one ctlsock and one workersock */
 370   if ((workersock = socket(PF_INET, SOCK_STREAM, 0)) < 0) {
 371     confd_fatal("Failed to open workersocket\n");
 372   }
 373
 374   if (confd_connect(dctx, workersock, WORKER_SOCKET,(struct sockaddr*)&addr,
 375                     sizeof (struct sockaddr_in)) < 0) {
 376     confd_fatal("Failed to confd_connect() to confd \n");
 377   }
 378
 379   if (confd_register_trans_cb(dctx, &trans) == CONFD_ERR) {
 380     confd_fatal("Failed to register trans cb \n");
 381   }
 382
 383   if (confd_register_data_cb(dctx, &data) == CONFD_ERR) {
 384     confd_fatal("Failed to register data cb \n");
 385   }
 386
 387   if (confd_register_action_cbs(dctx, &action) == CONFD_ERR) {
 388     confd_fatal("Failed to register action cb \n");
 389   }
 390
 391   if (confd_register_done(dctx) != CONFD_OK) {
 392     confd_fatal("Failed to complete registration \n");
 393   }
 394
 395   while(1) {
 396     struct pollfd set[3];
 397     int ret;
 398     set[0].fd = ctlsock;
 399     set[0].events = POLLIN;
 400     set[0].revents = 0;
 401     set[1].fd = workersock;
 402     set[1].events = POLLIN;
 403     set[1].revents = 0;
 404     set[2].fd = subsock;
 405     set[2].events = POLLIN;
 406     set[2].revents = 0;
 407     if (poll(set, sizeof(set)/sizeof(*set), -1) < 0) {
 408       perror("Poll failed:");
 409       continue;
 410     }
 411     /* Check for I/O */
 412     if (set[0].revents & POLLIN) {
 413       if ((ret = confd_fd_ready(dctx, ctlsock)) == CONFD_EOF) {
 414         confd_fatal("Control socket closed\n");
 415       } else if (ret == CONFD_ERR && confd_errno != CONFD_ERR_EXTERNAL) {
 416         confd_fatal("Error on control socket request: %s (%d): %s\n",
 417                     confd_strerror(confd_errno), confd_errno, confd_lasterr());
 418       }
 419     }
 420     if (set[1].revents & POLLIN) {
 421       if ((ret = confd_fd_ready(dctx, workersock)) == CONFD_EOF) {
 422         confd_fatal("Worker socket closed\n");
 423       } else if (ret == CONFD_ERR && confd_errno != CONFD_ERR_EXTERNAL) {
 424         confd_fatal("Error on worker socket request: %s (%d): %s\n",
 425                     confd_strerror(confd_errno), confd_errno, confd_lasterr());
 426       }
 427     }
 428     if (set[2].revents & POLLIN) {
 429       process_confd_subscription(set[2].fd);
 430     }
 431   }
 432
 433   return 0;
 434 }