src/amqp1.c

   1 /**
   2  * collectd - src/amqp1.c
   3  * Copyright(c) 2017 Red Hat Inc.
   4  *
   5  * Permission is hereby granted, free of charge, to any person obtaining a
   6  * copy of this software and associated documentation files (the "Software"),
   7  * to deal in the Software without restriction, including without limitation
   8  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9  * and/or sell copies of the Software, and to permit persons to whom the
  10  * Software is furnished to do so, subject to the following conditions:
  11  *
  12  * The above copyright notice and this permission notice shall be included in
  13  * all copies or substantial portions of the Software.
  14  *
  15  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  16  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  17  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  18  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  19  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  20  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  21  * DEALINGS IN THE SOFTWARE.
  22  *
  23  * Authors:
  24  *   Andy Smith <ansmith@redhat.com>
  25  */
  26
  27 #include "collectd.h"
  28
  29 #include "common.h"
  30 #include "plugin.h"
  31 #include "utils_cmd_putval.h"
  32 #include "utils_deq.h"
  33 #include "utils_format_graphite.h"
  34 #include "utils_format_json.h"
  35 #include "utils_random.h"
  36
  37 #include <proton/condition.h>
  38 #include <proton/connection.h>
  39 #include <proton/delivery.h>
  40 #include <proton/link.h>
  41 #include <proton/message.h>
  42 #include <proton/proactor.h>
  43 #include <proton/sasl.h>
  44 #include <proton/session.h>
  45 #include <proton/transport.h>
  46
  47 #include <errno.h>
  48 #include <stdint.h>
  49 #include <stdio.h>
  50 #include <stdlib.h>
  51
  52 #define BUFSIZE 8192
  53 #define AMQP1_FORMAT_JSON 0
  54 #define AMQP1_FORMAT_COMMAND 1
  55 #define AMQP1_FORMAT_GRAPHITE 2
  56
  57 typedef struct amqp1_config_transport_t {
  58   DEQ_LINKS(struct amqp1_config_transport_t);
  59   char *name;
  60   char *host;
  61   char *port;
  62   char *user;
  63   char *password;
  64   char *address;
  65   int retry_delay;
  66 } amqp1_config_transport_t;
  67
  68 typedef struct amqp1_config_instance_t {
  69   DEQ_LINKS(struct amqp1_config_instance_t);
  70   char *name;
  71   _Bool notify;
  72   uint8_t format;
  73   unsigned int graphite_flags;
  74   _Bool store_rates;
  75   char *prefix;
  76   char *postfix;
  77   char escape_char;
  78   _Bool pre_settle;
  79   char send_to[1024];
  80 } amqp1_config_instance_t;
  81
  82 DEQ_DECLARE(amqp1_config_instance_t, amqp1_config_instance_list_t);
  83
  84 typedef struct cd_message_t {
  85   DEQ_LINKS(struct cd_message_t);
  86   pn_rwbytes_t mbuf;
  87   amqp1_config_instance_t *instance;
  88 } cd_message_t;
  89
  90 DEQ_DECLARE(cd_message_t, cd_message_list_t);
  91
  92 /*
  93  * Globals
  94  */
  95 pn_connection_t *conn = NULL;
  96 pn_link_t *sender = NULL;
  97 pn_proactor_t *proactor = NULL;
  98 pthread_mutex_t send_lock;
  99 cd_message_list_t out_messages;
 100 uint64_t cd_tag = 1;
 101 uint64_t acknowledged = 0;
 102 amqp1_config_transport_t *transport = NULL;
 103
 104 static bool stopping = false;
 105 static int event_thread_running = 0;
 106 static pthread_t event_thread_id;
 107
 108 /*
 109  * Functions
 110  */
 111 static void cd_message_free(cd_message_t *cdm) {
 112   if (cdm->mbuf.start) {
 113     free((void *)cdm->mbuf.start);
 114   }
 115   free(cdm);
 116 } /* }}} void cd_message_free */
 117
 118 static int amqp1_send_out_messages(pn_link_t *link) /* {{{ */
 119 {
 120   uint64_t dtag;
 121   cd_message_list_t to_send;
 122   cd_message_t *cdm;
 123   int link_credit = pn_link_credit(link);
 124   int event_count = 0;
 125   pn_delivery_t *dlv;
 126
 127   if (stopping) {
 128     return 0;
 129   }
 130
 131   DEQ_INIT(to_send);
 132
 133   pthread_mutex_lock(&send_lock);
 134
 135   if (link_credit > 0) {
 136     dtag = cd_tag;
 137     cdm = DEQ_HEAD(out_messages);
 138     while (cdm) {
 139       DEQ_REMOVE_HEAD(out_messages);
 140       DEQ_INSERT_TAIL(to_send, cdm);
 141       if (DEQ_SIZE(to_send) == link_credit)
 142         break;
 143       cdm = DEQ_HEAD(out_messages);
 144     }
 145     cd_tag += DEQ_SIZE(to_send);
 146   }
 147
 148   pthread_mutex_unlock(&send_lock);
 149
 150   /* message is already formatted and encoded */
 151   cdm = DEQ_HEAD(to_send);
 152   while (cdm) {
 153     DEQ_REMOVE_HEAD(to_send);
 154     dtag++;
 155     dlv = pn_delivery(link, pn_dtag((const char *)&dtag, sizeof(dtag)));
 156     pn_link_send(link, cdm->mbuf.start, cdm->mbuf.size);
 157     pn_link_advance(link);
 158     if (cdm->instance->pre_settle == true) {
 159       pn_delivery_settle(dlv);
 160     }
 161     event_count++;
 162     cd_message_free(cdm);
 163     cdm = DEQ_HEAD(to_send);
 164   }
 165
 166   return event_count;
 167 } /* }}} int amqp1_send_out_messages */
 168
 169 static void check_condition(pn_event_t *e, pn_condition_t *cond) /* {{{ */
 170 {
 171   if (pn_condition_is_set(cond)) {
 172     ERROR("amqp1 plugin: %s: %s: %s", pn_event_type_name(pn_event_type(e)),
 173           pn_condition_get_name(cond), pn_condition_get_description(cond));
 174     pn_connection_close(pn_event_connection(e));
 175     conn = NULL;
 176   }
 177 } /* }}} void check_condition */
 178
 179 static bool handle(pn_event_t *event) /* {{{ */
 180 {
 181
 182   switch (pn_event_type(event)) {
 183
 184   case PN_CONNECTION_INIT: {
 185     conn = pn_event_connection(event);
 186     pn_connection_set_container(conn, transport->name);
 187     pn_connection_open(conn);
 188     pn_session_t *ssn = pn_session(conn);
 189     pn_session_open(ssn);
 190     sender = pn_sender(ssn, "cd-sender");
 191     pn_link_set_snd_settle_mode(sender, PN_SND_MIXED);
 192     pn_link_open(sender);
 193     break;
 194   }
 195
 196   case PN_LINK_FLOW: {
 197     /* peer has given us credit, send outbound messages */
 198     amqp1_send_out_messages(sender);
 199     break;
 200   }
 201
 202   case PN_DELIVERY: {
 203     /* acknowledgement from peer that a message was delivered */
 204     pn_delivery_t *dlv = pn_event_delivery(event);
 205     if (pn_delivery_remote_state(dlv) == PN_ACCEPTED) {
 206       pn_delivery_settle(dlv);
 207       acknowledged++;
 208     }
 209     break;
 210   }
 211
 212   case PN_CONNECTION_WAKE: {
 213     if (!stopping) {
 214       amqp1_send_out_messages(sender);
 215     }
 216     break;
 217   }
 218
 219   case PN_TRANSPORT_CLOSED: {
 220     check_condition(event, pn_transport_condition(pn_event_transport(event)));
 221     break;
 222   }
 223
 224   case PN_CONNECTION_REMOTE_CLOSE: {
 225     check_condition(event,
 226                     pn_session_remote_condition(pn_event_session(event)));
 227     pn_connection_close(pn_event_connection(event));
 228     break;
 229   }
 230
 231   case PN_SESSION_REMOTE_CLOSE: {
 232     check_condition(event,
 233                     pn_session_remote_condition(pn_event_session(event)));
 234     pn_connection_close(pn_event_connection(event));
 235     break;
 236   }
 237
 238   case PN_LINK_REMOTE_CLOSE:
 239   case PN_LINK_REMOTE_DETACH: {
 240     check_condition(event, pn_link_remote_condition(pn_event_link(event)));
 241     pn_connection_close(pn_event_connection(event));
 242     break;
 243   }
 244
 245   case PN_PROACTOR_INACTIVE: {
 246     return false;
 247   }
 248
 249   default:
 250     break;
 251   }
 252   return true;
 253 } /* }}} bool handle */
 254
 255 static void *event_thread(void __attribute__((unused)) * arg) /* {{{ */
 256 {
 257   char addr[PN_MAX_ADDR];
 258   cd_message_t *cdm;
 259
 260   /* setup proactor */
 261   proactor = pn_proactor();
 262   pn_proactor_addr(addr, sizeof(addr), transport->host, transport->port);
 263
 264   while (!stopping) {
 265     /* make connection */
 266     conn = pn_connection();
 267     if (transport->user != NULL) {
 268       pn_connection_set_user(conn, transport->user);
 269       pn_connection_set_password(conn, transport->password);
 270     }
 271     pn_proactor_connect(proactor, conn, addr);
 272
 273     bool engine_running = true;
 274     while (engine_running && !stopping) {
 275       pn_event_batch_t *events = pn_proactor_wait(proactor);
 276       pn_event_t *e;
 277       while ((e = pn_event_batch_next(events))) {
 278         engine_running = handle(e);
 279         if (!engine_running) {
 280           break;
 281         }
 282       }
 283       pn_proactor_done(proactor, events);
 284     }
 285
 286     pn_proactor_release_connection(conn);
 287
 288     DEBUG("amqp1 plugin: retrying connection");
 289     int delay = transport->retry_delay;
 290     while (delay-- > 0 && !stopping) {
 291       sleep(1.0);
 292     }
 293   }
 294
 295   pn_proactor_disconnect(proactor, NULL);
 296
 297   /* Free the remaining out_messages */
 298   cdm = DEQ_HEAD(out_messages);
 299   while (cdm) {
 300     DEQ_REMOVE_HEAD(out_messages);
 301     cd_message_free(cdm);
 302     cdm = DEQ_HEAD(out_messages);
 303   }
 304
 305   event_thread_running = 0;
 306
 307   return NULL;
 308 } /* }}} void event_thread */
 309
 310 static int encqueue(cd_message_t *cdm,
 311                     amqp1_config_instance_t *instance) /* {{{ */
 312 {
 313   size_t bufsize = BUFSIZE;
 314   pn_data_t *body;
 315   pn_message_t *message;
 316   int status = 0;
 317
 318   /* encode message */
 319   message = pn_message();
 320   pn_message_set_address(message, instance->send_to);
 321   body = pn_message_body(message);
 322   pn_data_clear(body);
 323   pn_data_put_binary(body, pn_bytes(cdm->mbuf.size, cdm->mbuf.start));
 324   pn_data_exit(body);
 325
 326   /* put_binary copies and stores so ok to use mbuf */
 327   cdm->mbuf.size = bufsize;
 328   while ((status = pn_message_encode(message, (char *)cdm->mbuf.start,
 329                                      &cdm->mbuf.size)) == PN_OVERFLOW) {
 330     DEBUG("amqp1 plugin: increasing message buffer size %i",
 331           (int)cdm->mbuf.size);
 332     cdm->mbuf.size *= 2;
 333     cdm->mbuf.start = (char *)realloc(cdm->mbuf.start, cdm->mbuf.size);
 334   }
 335
 336   if (status != 0) {
 337     ERROR("amqp1 plugin: error encoding message: %s",
 338           pn_error_text(pn_message_error(message)));
 339     pn_message_free(message);
 340     cd_message_free(cdm);
 341     return -1;
 342   }
 343
 344   pthread_mutex_lock(&send_lock);
 345   DEQ_INSERT_TAIL(out_messages, cdm);
 346   pthread_mutex_unlock(&send_lock);
 347
 348   pn_message_free(message);
 349
 350   /* activate the sender */
 351   if (conn != NULL) {
 352     pn_connection_wake(conn);
 353   }
 354
 355   return 0;
 356 } /* }}} int encqueue */
 357
 358 static int amqp1_notify(notification_t const *n,
 359                         user_data_t *user_data) /* {{{ */
 360 {
 361   amqp1_config_instance_t *instance;
 362   int status = 0;
 363   size_t bfree = BUFSIZE;
 364   size_t bfill = 0;
 365   cd_message_t *cdm;
 366   size_t bufsize = BUFSIZE;
 367
 368   if ((n == NULL) || (user_data == NULL))
 369     return EINVAL;
 370
 371   instance = user_data->data;
 372
 373   if (instance->notify != true) {
 374     ERROR("amqp1 plugin: write notification failed");
 375   }
 376
 377   cdm = NEW(cd_message_t);
 378   DEQ_ITEM_INIT(cdm);
 379   cdm->mbuf = pn_rwbytes(bufsize, (char *)malloc(bufsize));
 380   cdm->instance = instance;
 381
 382   switch (instance->format) {
 383   case AMQP1_FORMAT_JSON:
 384     format_json_initialize((char *)cdm->mbuf.start, &bfill, &bfree);
 385     status = format_json_notification((char *)cdm->mbuf.start, bufsize, n);
 386     if (status != 0) {
 387       ERROR("amqp1 plugin: formatting notification failed");
 388       return status;
 389     }
 390     cdm->mbuf.size = strlen(cdm->mbuf.start);
 391     break;
 392   default:
 393     ERROR("amqp1 plugin: Invalid notify format (%i).", instance->format);
 394     return -1;
 395   }
 396
 397   /* encode message and place on outbound queue */
 398   status = encqueue(cdm, instance);
 399
 400   return status;
 401 } /* }}} int amqp1_notify */
 402
 403 static int amqp1_write(const data_set_t *ds, const value_list_t *vl, /* {{{ */
 404                        user_data_t *user_data) {
 405   amqp1_config_instance_t *instance;
 406   int status = 0;
 407   size_t bfree = BUFSIZE;
 408   size_t bfill = 0;
 409   cd_message_t *cdm;
 410   size_t bufsize = BUFSIZE;
 411
 412   if ((ds == NULL) || (vl == NULL) || (transport == NULL) ||
 413       (user_data == NULL))
 414     return EINVAL;
 415
 416   instance = user_data->data;
 417
 418   if (instance->notify != false) {
 419     ERROR("amqp1 plugin: write failed");
 420   }
 421
 422   cdm = NEW(cd_message_t);
 423   DEQ_ITEM_INIT(cdm);
 424   cdm->mbuf = pn_rwbytes(bufsize, (char *)malloc(bufsize));
 425   cdm->instance = instance;
 426
 427   switch (instance->format) {
 428   case AMQP1_FORMAT_COMMAND:
 429     status = cmd_create_putval((char *)cdm->mbuf.start, bufsize, ds, vl);
 430     if (status != 0) {
 431       ERROR("amqp1 plugin: cmd_create_putval failed with status %i.", status);
 432       return status;
 433     }
 434     cdm->mbuf.size = strlen(cdm->mbuf.start);
 435     break;
 436   case AMQP1_FORMAT_JSON:
 437     format_json_initialize((char *)cdm->mbuf.start, &bfill, &bfree);
 438     format_json_value_list((char *)cdm->mbuf.start, &bfill, &bfree, ds, vl,
 439                            instance->store_rates);
 440     format_json_finalize((char *)cdm->mbuf.start, &bfill, &bfree);
 441     cdm->mbuf.size = strlen(cdm->mbuf.start);
 442     break;
 443   case AMQP1_FORMAT_GRAPHITE:
 444     status = format_graphite((char *)cdm->mbuf.start, bufsize, ds, vl,
 445                              instance->prefix, instance->postfix,
 446                              instance->escape_char, instance->graphite_flags);
 447     if (status != 0) {
 448       ERROR("amqp1 plugin: format_graphite failed with status %i.", status);
 449       return status;
 450     }
 451     cdm->mbuf.size = strlen(cdm->mbuf.start);
 452     break;
 453   default:
 454     ERROR("amqp1 plugin: Invalid write format (%i).", instance->format);
 455     return -1;
 456   }
 457
 458   /* encode message and place on outboud queue */
 459   encqueue(cdm, instance);
 460
 461   return 0;
 462 } /* }}} int amqp1_write */
 463
 464 static void amqp1_config_transport_free(void *ptr) /* {{{ */
 465 {
 466   amqp1_config_transport_t *transport = ptr;
 467
 468   if (transport == NULL)
 469     return;
 470
 471   sfree(transport->name);
 472   sfree(transport->host);
 473   sfree(transport->user);
 474   sfree(transport->password);
 475   sfree(transport->address);
 476
 477   sfree(transport);
 478 } /* }}} void amqp1_config_transport_free */
 479
 480 static void amqp1_config_instance_free(void *ptr) /* {{{ */
 481 {
 482   amqp1_config_instance_t *instance = ptr;
 483
 484   if (instance == NULL)
 485     return;
 486
 487   sfree(instance->name);
 488   sfree(instance->prefix);
 489   sfree(instance->postfix);
 490
 491   sfree(instance);
 492 } /* }}} void amqp1_config_instance_free */
 493
 494 static int amqp1_config_instance(oconfig_item_t *ci) /* {{{ */
 495 {
 496   int status = 0;
 497   char *key = NULL;
 498   amqp1_config_instance_t *instance;
 499
 500   instance = calloc(1, sizeof(*instance));
 501   if (instance == NULL) {
 502     ERROR("amqp1 plugin: calloc failed.");
 503     return ENOMEM;
 504   }
 505
 506   /* Initialize instance configuration {{{ */
 507   instance->name = NULL;
 508
 509   status = cf_util_get_string(ci, &instance->name);
 510   if (status != 0) {
 511     sfree(instance);
 512     return status;
 513   }
 514
 515   for (int i = 0; i < ci->children_num; i++) {
 516     oconfig_item_t *child = ci->children + i;
 517
 518     if (strcasecmp("PreSettle", child->key) == 0)
 519       status = cf_util_get_boolean(child, &instance->pre_settle);
 520     else if (strcasecmp("Notify", child->key) == 0)
 521       status = cf_util_get_boolean(child, &instance->notify);
 522     else if (strcasecmp("Format", child->key) == 0) {
 523       status = cf_util_get_string(child, &key);
 524       if (status != 0)
 525         return status;
 526       /* TODO: goto errout */
 527       //          goto errout;
 528       assert(key != NULL);
 529       if (strcasecmp(key, "Command") == 0) {
 530         instance->format = AMQP1_FORMAT_COMMAND;
 531       } else if (strcasecmp(key, "Graphite") == 0) {
 532         instance->format = AMQP1_FORMAT_GRAPHITE;
 533       } else if (strcasecmp(key, "JSON") == 0) {
 534         instance->format = AMQP1_FORMAT_JSON;
 535       } else {
 536         WARNING("amqp1 plugin: Invalid format string: %s", key);
 537       }
 538       sfree(key);
 539     } else if (strcasecmp("StoreRates", child->key) == 0)
 540       status = cf_util_get_boolean(child, &instance->store_rates);
 541     else if (strcasecmp("GraphiteSeparateInstances", child->key) == 0)
 542       status = cf_util_get_flag(child, &instance->graphite_flags,
 543                                 GRAPHITE_SEPARATE_INSTANCES);
 544     else if (strcasecmp("GraphiteAlwaysAppendDS", child->key) == 0)
 545       status = cf_util_get_flag(child, &instance->graphite_flags,
 546                                 GRAPHITE_ALWAYS_APPEND_DS);
 547     else if (strcasecmp("GraphitePreserveSeparator", child->key) == 0)
 548       status = cf_util_get_flag(child, &instance->graphite_flags,
 549                                 GRAPHITE_PRESERVE_SEPARATOR);
 550     else if (strcasecmp("GraphitePrefix", child->key) == 0)
 551       status = cf_util_get_string(child, &instance->prefix);
 552     else if (strcasecmp("GraphitePostfix", child->key) == 0)
 553       status = cf_util_get_string(child, &instance->postfix);
 554     else if (strcasecmp("GraphiteEscapeChar", child->key) == 0) {
 555       char *tmp_buff = NULL;
 556       status = cf_util_get_string(child, &tmp_buff);
 557       if (strlen(tmp_buff) > 1)
 558         WARNING("amqp1 plugin: The option \"GraphiteEscapeChar\" handles "
 559                 "only one character. Others will be ignored.");
 560       instance->escape_char = tmp_buff[0];
 561       sfree(tmp_buff);
 562     } else
 563       WARNING("amqp1 plugin: Ignoring unknown "
 564               "instance configuration option "
 565               "\%s\".",
 566               child->key);
 567     if (status != 0)
 568       break;
 569   }
 570
 571   if (status != 0) {
 572     amqp1_config_instance_free(instance);
 573     return status;
 574   } else {
 575     char tpname[1024];
 576     int status;
 577     status = snprintf(tpname, sizeof(tpname), "amqp1/%s", instance->name);
 578     if ((status < 0) || (size_t)status >= sizeof(tpname)) {
 579       ERROR("amqp1 plugin: Instance name would have been truncated.");
 580       return -1;
 581     }
 582     status = snprintf(instance->send_to, sizeof(instance->send_to), "/%s/%s",
 583                       transport->address, instance->name);
 584     if ((status < 0) || (size_t)status >= sizeof(instance->send_to)) {
 585       ERROR("amqp1 plugin: send_to address would have been truncated.");
 586       return -1;
 587     }
 588     if (instance->notify == true) {
 589       status = plugin_register_notification(
 590           tpname, amqp1_notify,
 591           &(user_data_t){
 592               .data = instance, .free_func = amqp1_config_instance_free,
 593           });
 594     } else {
 595       status = plugin_register_write(
 596           tpname, amqp1_write,
 597           &(user_data_t){
 598               .data = instance, .free_func = amqp1_config_instance_free,
 599           });
 600     }
 601
 602     if (status != 0) {
 603       amqp1_config_instance_free(instance);
 604     }
 605   }
 606
 607   return status;
 608 } /* }}} int amqp1_config_instance */
 609
 610 static int amqp1_config_transport(oconfig_item_t *ci) /* {{{ */
 611 {
 612   int status = 0;
 613
 614   transport = calloc(1, sizeof(*transport));
 615   if (transport == NULL) {
 616     ERROR("amqp1 plugin: calloc failed.");
 617     return ENOMEM;
 618   }
 619
 620   /* Initialize transport configuration {{{ */
 621   transport->name = NULL;
 622   transport->retry_delay = 1;
 623
 624   status = cf_util_get_string(ci, &transport->name);
 625   if (status != 0) {
 626     sfree(transport);
 627     return status;
 628   }
 629
 630   for (int i = 0; i < ci->children_num; i++) {
 631     oconfig_item_t *child = ci->children + i;
 632
 633     if (strcasecmp("Host", child->key) == 0)
 634       status = cf_util_get_string(child, &transport->host);
 635     else if (strcasecmp("Port", child->key) == 0)
 636       status = cf_util_get_string(child, &transport->port);
 637     else if (strcasecmp("User", child->key) == 0)
 638       status = cf_util_get_string(child, &transport->user);
 639     else if (strcasecmp("Password", child->key) == 0)
 640       status = cf_util_get_string(child, &transport->password);
 641     else if (strcasecmp("Address", child->key) == 0)
 642       status = cf_util_get_string(child, &transport->address);
 643     else if (strcasecmp("RetryDelay", child->key) == 0)
 644       status = cf_util_get_int(child, &transport->retry_delay);
 645     else if (strcasecmp("Instance", child->key) == 0)
 646       amqp1_config_instance(child);
 647     else
 648       WARNING("amqp1 plugin: Ignoring unknown "
 649               "transport configuration option "
 650               "\%s\".",
 651               child->key);
 652
 653     if (status != 0)
 654       break;
 655   }
 656
 657   if (status != 0) {
 658     amqp1_config_transport_free(transport);
 659   }
 660   return status;
 661 } /* }}} int amqp1_config_transport */
 662
 663 static int amqp1_config(oconfig_item_t *ci) /* {{{ */
 664 {
 665
 666   for (int i = 0; i < ci->children_num; i++) {
 667     oconfig_item_t *child = ci->children + i;
 668
 669     if (strcasecmp("Transport", child->key) == 0)
 670       amqp1_config_transport(child);
 671     else
 672       WARNING("amqp1 plugin: Ignoring unknown config iption \%s\".",
 673               child->key);
 674   }
 675
 676   return 0;
 677 } /* }}} int amqp1_config */
 678
 679 static int amqp1_init(void) /* {{{ */
 680 {
 681   int status;
 682   char errbuf[1024];
 683
 684   if (transport == NULL) {
 685     ERROR("amqp1: init failed, no transport configured");
 686     return -1;
 687   }
 688
 689   if (proactor == NULL) {
 690     pthread_mutex_init(&send_lock, /* attr = */ NULL);
 691     /* start_thread */
 692     status =
 693         plugin_thread_create(&event_thread_id, NULL /* no attributes */,
 694                              event_thread, NULL /* no argument */, "handle");
 695     if (status != 0) {
 696       ERROR("amqp1 plugin: pthread_create failed: %s",
 697             sstrerror(errno, errbuf, sizeof(errbuf)));
 698     } else {
 699       event_thread_running = 1;
 700     }
 701   }
 702   return 0;
 703 } /* }}} int amqp1_init */
 704
 705 static int amqp1_shutdown(void) /* {{{ */
 706 {
 707   stopping = true;
 708
 709   /* Stop the proactor thread */
 710   if (event_thread_running == 1) {
 711     DEBUG("amqp1 plugin: Shutting down proactor thread.");
 712     pn_connection_wake(conn);
 713   }
 714   pthread_join(event_thread_id, NULL /* no return value */);
 715   memset(&event_thread_id, 0, sizeof(event_thread_id));
 716
 717   DEBUG("amqp1 plugin: proactor thread exited.");
 718
 719   if (transport != NULL) {
 720     amqp1_config_transport_free(transport);
 721   }
 722
 723   return 0;
 724 } /* }}} int amqp1_shutdown */
 725
 726 void module_register(void) {
 727   plugin_register_complex_config("amqp1", amqp1_config);
 728   plugin_register_init("amqp1", amqp1_init);
 729   plugin_register_shutdown("amqp1", amqp1_shutdown);
 730 } /* void module_register */