src/amqp.c

   1 /**
   2  * collectd - src/amqp.c
   3  * Copyright (C) 2009  Sebastien Pahl
   4  * Copyright (C) 2010  Florian Forster
   5  *
   6  * Permission is hereby granted, free of charge, to any person obtaining a
   7  * copy of this software and associated documentation files (the "Software"),
   8  * to deal in the Software without restriction, including without limitation
   9  * the rights to use, copy, modify, merge, publish, distribute, sublicense,
  10  * and/or sell copies of the Software, and to permit persons to whom the
  11  * Software is furnished to do so, subject to the following conditions:
  12  *
  13  * The above copyright notice and this permission notice shall be included in
  14  * all copies or substantial portions of the Software.
  15  *
  16  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  19  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  21  * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
  22  * DEALINGS IN THE SOFTWARE.
  23  *
  24  * Authors:
  25  *   Sebastien Pahl <sebastien.pahl at dotcloud.com>
  26  *   Florian Forster <octo at verplant.org>
  27  **/
  28
  29 #include <stdint.h>
  30 #include <stdlib.h>
  31 #include <unistd.h>
  32 #include <strings.h>
  33 #include <pthread.h>
  34
  35 #include "collectd.h"
  36 #include "common.h"
  37 #include "plugin.h"
  38 #include "utils_format_json.h"
  39
  40 #include <amqp.h>
  41 #include <amqp_framing.h>
  42
  43 /* Defines for the delivery mode. I have no idea why they're not defined by the
  44  * library.. */
  45 #define CAMQP_DM_VOLATILE   1
  46 #define CAMQP_DM_PERSISTENT 2
  47
  48 #define CAMQP_CHANNEL 1
  49
  50 /*
  51  * Data types
  52  */
  53 struct camqp_config_s
  54 {
  55     _Bool   publish;
  56     char   *name;
  57
  58     char   *host;
  59     int     port;
  60     char   *vhost;
  61     char   *user;
  62     char   *password;
  63
  64     char   *exchange;
  65     char   *exchange_type;
  66     char   *queue;
  67     char   *routingkey;
  68     uint8_t delivery_mode;
  69
  70     _Bool   store_rates;
  71
  72     amqp_connection_state_t connection;
  73     pthread_mutex_t lock;
  74 };
  75 typedef struct camqp_config_s camqp_config_t;
  76
  77 /*
  78  * Global variables
  79  */
  80 static const char *def_host       = "localhost";
  81 static const char *def_vhost      = "/";
  82 static const char *def_user       = "guest";
  83 static const char *def_password   = "guest";
  84 static const char *def_exchange   = "amq.fanout";
  85 static const char *def_routingkey = "collectd";
  86
  87 static pthread_t *subscriber_threads     = NULL;
  88 static size_t     subscriber_threads_num = 0;
  89 static _Bool      subscriber_threads_running = 1;
  90
  91 #define CONF(c,f) (((c)->f != NULL) ? (c)->f : def_##f)
  92
  93 /*
  94  * Functions
  95  */
  96 static void camqp_close_connection (camqp_config_t *conf) /* {{{ */
  97 {
  98     int sockfd;
  99
 100     if ((conf == NULL) || (conf->connection == NULL))
 101         return;
 102
 103     sockfd = amqp_get_sockfd (conf->connection);
 104     amqp_channel_close (conf->connection, CAMQP_CHANNEL, AMQP_REPLY_SUCCESS);
 105     amqp_connection_close (conf->connection, AMQP_REPLY_SUCCESS);
 106     amqp_destroy_connection (conf->connection);
 107     close (sockfd);
 108     conf->connection = NULL;
 109 } /* }}} void camqp_close_connection */
 110
 111 static void camqp_config_free (void *ptr) /* {{{ */
 112 {
 113     camqp_config_t *conf = ptr;
 114
 115     if (conf == NULL)
 116         return;
 117
 118     camqp_close_connection (conf);
 119
 120     sfree (conf->name);
 121     sfree (conf->host);
 122     sfree (conf->vhost);
 123     sfree (conf->user);
 124     sfree (conf->password);
 125     sfree (conf->exchange);
 126     sfree (conf->exchange_type);
 127     sfree (conf->queue);
 128     sfree (conf->routingkey);
 129
 130     sfree (conf);
 131 } /* }}} void camqp_config_free */
 132
 133 static char *camqp_bytes_cstring (amqp_bytes_t *in) /* {{{ */
 134 {
 135     char *ret;
 136
 137     if ((in == NULL) || (in->bytes == NULL))
 138         return (NULL);
 139
 140     ret = malloc (in->len + 1);
 141     if (ret == NULL)
 142         return (NULL);
 143
 144     memcpy (ret, in->bytes, in->len);
 145     ret[in->len] = 0;
 146
 147     return (ret);
 148 } /* }}} char *camqp_bytes_cstring */
 149
 150 static _Bool camqp_is_error (camqp_config_t *conf) /* {{{ */
 151 {
 152     amqp_rpc_reply_t r;
 153
 154     r = amqp_get_rpc_reply (conf->connection);
 155     if (r.reply_type == AMQP_RESPONSE_NORMAL)
 156         return (0);
 157
 158     return (1);
 159 } /* }}} _Bool camqp_is_error */
 160
 161 static char *camqp_strerror (camqp_config_t *conf, /* {{{ */
 162         char *buffer, size_t buffer_size)
 163 {
 164     amqp_rpc_reply_t r;
 165
 166     r = amqp_get_rpc_reply (conf->connection);
 167     switch (r.reply_type)
 168     {
 169         case AMQP_RESPONSE_NORMAL:
 170             sstrncpy (buffer, "Success", sizeof (buffer));
 171             break;
 172
 173         case AMQP_RESPONSE_NONE:
 174             sstrncpy (buffer, "Missing RPC reply type", sizeof (buffer));
 175             break;
 176
 177         case AMQP_RESPONSE_LIBRARY_EXCEPTION:
 178             if (r.library_errno)
 179                 return (sstrerror (r.library_errno, buffer, buffer_size));
 180             else
 181                 sstrncpy (buffer, "End of stream", sizeof (buffer));
 182             break;
 183
 184         case AMQP_RESPONSE_SERVER_EXCEPTION:
 185             if (r.reply.id == AMQP_CONNECTION_CLOSE_METHOD)
 186             {
 187                 amqp_connection_close_t *m = r.reply.decoded;
 188                 char *tmp = camqp_bytes_cstring (&m->reply_text);
 189                 ssnprintf (buffer, buffer_size, "Server connection error %d: %s",
 190                         m->reply_code, tmp);
 191                 sfree (tmp);
 192             }
 193             else if (r.reply.id == AMQP_CHANNEL_CLOSE_METHOD)
 194             {
 195                 amqp_channel_close_t *m = r.reply.decoded;
 196                 char *tmp = camqp_bytes_cstring (&m->reply_text);
 197                 ssnprintf (buffer, buffer_size, "Server channel error %d: %s",
 198                         m->reply_code, tmp);
 199                 sfree (tmp);
 200             }
 201             else
 202             {
 203                 ssnprintf (buffer, buffer_size, "Server error method %#"PRIx32,
 204                         r.reply.id);
 205             }
 206             break;
 207
 208         default:
 209             ssnprintf (buffer, buffer_size, "Unknown reply type %i",
 210                     (int) r.reply_type);
 211     }
 212
 213     return (buffer);
 214 } /* }}} char *camqp_strerror */
 215
 216 static int camqp_setup_queue (camqp_config_t *conf) /* {{{ */
 217 {
 218     amqp_queue_declare_ok_t *qd_ret;
 219     amqp_basic_consume_ok_t *cm_ret;
 220
 221     qd_ret = amqp_queue_declare (conf->connection,
 222             /* channel     = */ CAMQP_CHANNEL,
 223             /* queue       = */ (conf->queue != NULL)
 224             ? amqp_cstring_bytes (conf->queue)
 225             : AMQP_EMPTY_BYTES,
 226             /* passive     = */ 0,
 227             /* durable     = */ 0,
 228             /* exclusive   = */ 0,
 229             /* auto_delete = */ 1,
 230             /* arguments   = */ AMQP_EMPTY_TABLE);
 231     if (qd_ret == NULL)
 232     {
 233         ERROR ("amqp plugin: amqp_queue_declare failed.");
 234         camqp_close_connection (conf);
 235         return (-1);
 236     }
 237
 238     if (conf->queue == NULL)
 239     {
 240         conf->queue = camqp_bytes_cstring (&qd_ret->queue);
 241         if (conf->queue == NULL)
 242         {
 243             ERROR ("amqp plugin: camqp_bytes_cstring failed.");
 244             camqp_close_connection (conf);
 245             return (-1);
 246         }
 247
 248         INFO ("amqp plugin: Created queue \"%s\".", conf->queue);
 249     }
 250     DEBUG ("amqp plugin: Successfully created queue \"%s\".", conf->queue);
 251
 252     /* bind to an exchange */
 253     if (conf->exchange != NULL)
 254     {
 255         amqp_queue_bind_ok_t *qb_ret;
 256
 257         /* create the exchange */
 258         if (conf->exchange_type != NULL)
 259         {
 260             amqp_exchange_declare_ok_t *ed_ret;
 261
 262             ed_ret = amqp_exchange_declare (conf->connection,
 263                     /* channel     = */ CAMQP_CHANNEL,
 264                     /* exchange    = */ amqp_cstring_bytes (conf->exchange),
 265                     /* type        = */ amqp_cstring_bytes (conf->exchange_type),
 266                     /* passive     = */ 0,
 267                     /* durable     = */ 0,
 268                     /* auto_delete = */ 1,
 269                     /* arguments   = */ AMQP_EMPTY_TABLE);
 270             if ((ed_ret == NULL) && camqp_is_error (conf))
 271             {
 272                 char errbuf[1024];
 273                 ERROR ("amqp plugin: amqp_exchange_declare failed: %s",
 274                         camqp_strerror (conf, errbuf, sizeof (errbuf)));
 275                 camqp_close_connection (conf);
 276                 return (-1);
 277             }
 278         }
 279
 280         DEBUG ("amqp plugin: queue = %s; exchange = %s; routing_key = %s;",
 281                 conf->queue, conf->exchange, CONF (conf, routingkey));
 282
 283         assert (conf->queue != NULL);
 284         qb_ret = amqp_queue_bind (conf->connection,
 285                 /* channel     = */ CAMQP_CHANNEL,
 286                 /* queue       = */ amqp_cstring_bytes (conf->queue),
 287                 /* exchange    = */ amqp_cstring_bytes (conf->exchange),
 288 #if 1
 289                 /* routing_key = */ amqp_cstring_bytes (CONF (conf, routingkey)),
 290 #else
 291                 /* routing_key = */ AMQP_EMPTY_BYTES,
 292 #endif
 293                 /* arguments   = */ AMQP_EMPTY_TABLE);
 294         if ((qb_ret == NULL) && camqp_is_error (conf))
 295         {
 296             char errbuf[1024];
 297             ERROR ("amqp plugin: amqp_queue_bind failed: %s",
 298                     camqp_strerror (conf, errbuf, sizeof (errbuf)));
 299             camqp_close_connection (conf);
 300             return (-1);
 301         }
 302     } /* if (conf->exchange != NULL) */
 303
 304     cm_ret = amqp_basic_consume (conf->connection,
 305             /* channel      = */ CAMQP_CHANNEL,
 306             /* queue        = */ amqp_cstring_bytes (conf->queue),
 307             /* consumer_tag = */ AMQP_EMPTY_BYTES,
 308             /* no_local     = */ 0,
 309             /* no_ack       = */ 1,
 310             /* exclusive    = */ 0);
 311     if ((cm_ret == NULL) && camqp_is_error (conf))
 312     {
 313         char errbuf[1024];
 314         ERROR ("amqp plugin: amqp_basic_consume failed: %s",
 315                     camqp_strerror (conf, errbuf, sizeof (errbuf)));
 316         camqp_close_connection (conf);
 317         return (-1);
 318     }
 319
 320     return (0);
 321 } /* }}} int camqp_setup_queue */
 322
 323 static int camqp_connect (camqp_config_t *conf) /* {{{ */
 324 {
 325     amqp_rpc_reply_t reply;
 326     int sockfd;
 327     int status;
 328
 329     if (conf->connection != NULL)
 330         return (0);
 331
 332     conf->connection = amqp_new_connection ();
 333     if (conf->connection == NULL)
 334     {
 335         ERROR ("amqp plugin: amqp_new_connection failed.");
 336         return (ENOMEM);
 337     }
 338
 339     sockfd = amqp_open_socket (CONF(conf, host), conf->port);
 340     if (sockfd < 0)
 341     {
 342         char errbuf[1024];
 343         status = (-1) * sockfd;
 344         ERROR ("amqp plugin: amqp_open_socket failed: %s",
 345                 sstrerror (status, errbuf, sizeof (errbuf)));
 346         amqp_destroy_connection (conf->connection);
 347         conf->connection = NULL;
 348         return (status);
 349     }
 350     amqp_set_sockfd (conf->connection, sockfd);
 351
 352     reply = amqp_login (conf->connection, CONF(conf, vhost),
 353             /* channel max = */      0,
 354             /* frame max   = */ 131072,
 355             /* heartbeat   = */      0,
 356             /* authentication = */ AMQP_SASL_METHOD_PLAIN,
 357             CONF(conf, user), CONF(conf, password));
 358     if (reply.reply_type != AMQP_RESPONSE_NORMAL)
 359     {
 360         ERROR ("amqp plugin: amqp_login (vhost = %s, user = %s) failed.",
 361                 CONF(conf, vhost), CONF(conf, user));
 362         amqp_destroy_connection (conf->connection);
 363         close (sockfd);
 364         conf->connection = NULL;
 365         return (1);
 366     }
 367
 368     amqp_channel_open (conf->connection, /* channel = */ 1);
 369     /* FIXME: Is checking "reply.reply_type" really correct here? How does
 370      * it get set? --octo */
 371     if (reply.reply_type != AMQP_RESPONSE_NORMAL)
 372     {
 373         ERROR ("amqp plugin: amqp_channel_open failed.");
 374         amqp_connection_close (conf->connection, AMQP_REPLY_SUCCESS);
 375         amqp_destroy_connection (conf->connection);
 376         close(sockfd);
 377         conf->connection = NULL;
 378         return (1);
 379     }
 380
 381     INFO ("amqp plugin: Successfully opened connection to vhost \"%s\" "
 382             "on %s:%i.", CONF(conf, vhost), CONF(conf, host), conf->port);
 383
 384     if (!conf->publish)
 385         return (camqp_setup_queue (conf));
 386     return (0);
 387 } /* }}} int camqp_connect */
 388
 389 static int camqp_read_body (camqp_config_t *conf, /* {{{ */
 390         size_t body_size)
 391 {
 392     char body[body_size + 1];
 393     char *body_ptr;
 394     size_t received;
 395     amqp_frame_t frame;
 396     int status;
 397
 398     memset (body, 0, sizeof (body));
 399     body_ptr = &body[0];
 400     received = 0;
 401
 402     while (received < body_size)
 403     {
 404         status = amqp_simple_wait_frame (conf->connection, &frame);
 405         if (status < 0)
 406         {
 407             char errbuf[1024];
 408             status = (-1) * status;
 409             ERROR ("amqp plugin: amqp_simple_wait_frame failed: %s",
 410                     sstrerror (status, errbuf, sizeof (errbuf)));
 411             camqp_close_connection (conf);
 412             return (status);
 413         }
 414
 415         if (frame.frame_type != AMQP_FRAME_BODY)
 416         {
 417             NOTICE ("amqp plugin: Unexpected frame type: %#"PRIx8,
 418                     frame.frame_type);
 419             return (-1);
 420         }
 421
 422         if ((body_size - received) < frame.payload.body_fragment.len)
 423         {
 424             WARNING ("amqp plugin: Body is larger than indicated by header.");
 425             return (-1);
 426         }
 427
 428         memcpy (body_ptr, frame.payload.body_fragment.bytes,
 429                 frame.payload.body_fragment.len);
 430         body_ptr += frame.payload.body_fragment.len;
 431         received += frame.payload.body_fragment.len;
 432     } /* while (received < body_size) */
 433
 434     DEBUG ("amqp plugin: camqp_read_body: body = %s", body);
 435
 436     return (0);
 437 } /* }}} int camqp_read_body */
 438
 439 static int camqp_read_header (camqp_config_t *conf) /* {{{ */
 440 {
 441     int status;
 442     amqp_frame_t frame;
 443
 444     status = amqp_simple_wait_frame (conf->connection, &frame);
 445     if (status < 0)
 446     {
 447         char errbuf[1024];
 448         status = (-1) * status;
 449         ERROR ("amqp plugin: amqp_simple_wait_frame failed: %s",
 450                     sstrerror (status, errbuf, sizeof (errbuf)));
 451         camqp_close_connection (conf);
 452         return (status);
 453     }
 454
 455     if (frame.frame_type != AMQP_FRAME_HEADER)
 456     {
 457         NOTICE ("amqp plugin: Unexpected frame type: %#"PRIx8,
 458                 frame.frame_type);
 459         return (-1);
 460     }
 461
 462     return (camqp_read_body (conf, frame.payload.properties.body_size));
 463 } /* }}} int camqp_read_header */
 464
 465 static void *camqp_subscribe_thread (void *user_data) /* {{{ */
 466 {
 467     camqp_config_t *conf = user_data;
 468     int status;
 469
 470     while (subscriber_threads_running)
 471     {
 472         amqp_frame_t frame;
 473
 474         status = camqp_connect (conf);
 475         if (status != 0)
 476         {
 477             ERROR ("amqp plugin: camqp_connect failed. "
 478                     "Will sleep for %i seconds.", interval_g);
 479             sleep (interval_g);
 480             continue;
 481         }
 482
 483         status = amqp_simple_wait_frame (conf->connection, &frame);
 484         if (status < 0)
 485         {
 486             ERROR ("amqp plugin: amqp_simple_wait_frame failed. "
 487                     "Will sleep for %i seconds.", interval_g);
 488             camqp_close_connection (conf);
 489             sleep (interval_g);
 490             continue;
 491         }
 492
 493         if (frame.frame_type != AMQP_FRAME_METHOD)
 494         {
 495             DEBUG ("amqp plugin: Unexpected frame type: %#"PRIx8,
 496                     frame.frame_type);
 497             continue;
 498         }
 499
 500         if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
 501         {
 502             DEBUG ("amqp plugin: Unexpected method id: %#"PRIx32,
 503                     frame.payload.method.id);
 504             continue;
 505         }
 506
 507         status = camqp_read_header (conf);
 508
 509         amqp_maybe_release_buffers (conf->connection);
 510     } /* while (subscriber_threads_running) */
 511
 512     camqp_config_free (conf);
 513     pthread_exit (NULL);
 514 } /* }}} void *camqp_subscribe_thread */
 515
 516 static int camqp_subscribe_init (camqp_config_t *conf) /* {{{ */
 517 {
 518     int status;
 519     pthread_t *tmp;
 520
 521     tmp = realloc (subscriber_threads,
 522             sizeof (*subscriber_threads) * (subscriber_threads_num + 1));
 523     if (tmp == NULL)
 524     {
 525         ERROR ("amqp plugin: realloc failed.");
 526         camqp_config_free (conf);
 527         return (ENOMEM);
 528     }
 529     subscriber_threads = tmp;
 530     tmp = subscriber_threads + subscriber_threads_num;
 531     memset (tmp, 0, sizeof (*tmp));
 532
 533     status = pthread_create (tmp, /* attr = */ NULL,
 534             camqp_subscribe_thread, conf);
 535     if (status != 0)
 536     {
 537         char errbuf[1024];
 538         ERROR ("amqp plugin: pthread_create failed: %s",
 539                 sstrerror (status, errbuf, sizeof (errbuf)));
 540         camqp_config_free (conf);
 541         return (status);
 542     }
 543
 544     subscriber_threads_num++;
 545
 546     return (0);
 547 } /* }}} int camqp_subscribe_init */
 548
 549 static int camqp_write_locked (camqp_config_t *conf, /* {{{ */
 550         const char *buffer)
 551 {
 552     amqp_basic_properties_t props;
 553     int status;
 554
 555     status = camqp_connect (conf);
 556     if (status != 0)
 557         return (status);
 558
 559     memset (&props, 0, sizeof (props));
 560     props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG
 561         | AMQP_BASIC_DELIVERY_MODE_FLAG
 562         | AMQP_BASIC_APP_ID_FLAG;
 563     props.content_type = amqp_cstring_bytes("application/json");
 564     props.delivery_mode = conf->delivery_mode;
 565     props.app_id = amqp_cstring_bytes("collectd");
 566
 567     status = amqp_basic_publish(conf->connection,
 568                 /* channel = */ 1,
 569                 amqp_cstring_bytes(CONF(conf, exchange)),
 570                 amqp_cstring_bytes(CONF(conf, routingkey)),
 571                 /* mandatory = */ 0,
 572                 /* immediate = */ 0,
 573                 &props,
 574                 amqp_cstring_bytes(buffer));
 575     if (status != 0)
 576     {
 577         ERROR ("amqp plugin: amqp_basic_publish failed with status %i.",
 578                 status);
 579         camqp_close_connection (conf);
 580     }
 581
 582     return (status);
 583 } /* }}} int camqp_write_locked */
 584
 585 static int camqp_write (const data_set_t *ds, const value_list_t *vl, /* {{{ */
 586         user_data_t *user_data)
 587 {
 588     camqp_config_t *conf = user_data->data;
 589     char buffer[4096];
 590     size_t bfree;
 591     size_t bfill;
 592     int status;
 593
 594     if ((ds == NULL) || (vl == NULL) || (conf == NULL))
 595         return (EINVAL);
 596
 597     memset (buffer, 0, sizeof (buffer));
 598     bfree = sizeof (buffer);
 599     bfill = 0;
 600
 601     format_json_initialize (buffer, &bfill, &bfree);
 602     format_json_value_list (buffer, &bfill, &bfree, ds, vl, conf->store_rates);
 603     format_json_finalize (buffer, &bfill, &bfree);
 604
 605     pthread_mutex_lock (&conf->lock);
 606     status = camqp_write_locked (conf, buffer);
 607     pthread_mutex_unlock (&conf->lock);
 608
 609     return (status);
 610 } /* }}} int camqp_write */
 611
 612 static int camqp_config_connection (oconfig_item_t *ci, /* {{{ */
 613         _Bool publish)
 614 {
 615     camqp_config_t *conf;
 616     int status;
 617     int i;
 618
 619     conf = malloc (sizeof (*conf));
 620     if (conf == NULL)
 621     {
 622         ERROR ("amqp plugin: malloc failed.");
 623         return (ENOMEM);
 624     }
 625
 626     /* Initialize "conf" {{{ */
 627     memset (conf, 0, sizeof (*conf));
 628     conf->publish = publish;
 629     conf->name = NULL;
 630     conf->host = NULL;
 631     conf->port = 5672;
 632     conf->vhost = NULL;
 633     conf->user = NULL;
 634     conf->password = NULL;
 635     conf->exchange = NULL;
 636     conf->exchange_type = NULL;
 637     conf->queue = NULL;
 638     conf->routingkey = NULL;
 639     conf->delivery_mode = CAMQP_DM_VOLATILE;
 640     conf->store_rates = 0;
 641     conf->connection = NULL;
 642     pthread_mutex_init (&conf->lock, /* attr = */ NULL);
 643     /* }}} */
 644
 645     status = cf_util_get_string (ci, &conf->name);
 646     if (status != 0)
 647     {
 648         sfree (conf);
 649         return (status);
 650     }
 651
 652     for (i = 0; i < ci->children_num; i++)
 653     {
 654         oconfig_item_t *child = ci->children + i;
 655
 656         if (strcasecmp ("Host", child->key) == 0)
 657             status = cf_util_get_string (child, &conf->host);
 658         else if (strcasecmp ("Port", child->key) == 0)
 659         {
 660             status = cf_util_get_port_number (child);
 661             if (status > 0)
 662             {
 663                 conf->port = status;
 664                 status = 0;
 665             }
 666         }
 667         else if (strcasecmp ("VHost", child->key) == 0)
 668             status = cf_util_get_string (child, &conf->vhost);
 669         else if (strcasecmp ("User", child->key) == 0)
 670             status = cf_util_get_string (child, &conf->user);
 671         else if (strcasecmp ("Password", child->key) == 0)
 672             status = cf_util_get_string (child, &conf->password);
 673         else if (strcasecmp ("Exchange", child->key) == 0)
 674             status = cf_util_get_string (child, &conf->exchange);
 675         else if ((strcasecmp ("ExchangeType", child->key) == 0) && !publish)
 676             status = cf_util_get_string (child, &conf->exchange_type);
 677         else if ((strcasecmp ("Queue", child->key) == 0) && !publish)
 678             status = cf_util_get_string (child, &conf->queue);
 679         else if (strcasecmp ("RoutingKey", child->key) == 0)
 680             status = cf_util_get_string (child, &conf->routingkey);
 681         else if (strcasecmp ("Persistent", child->key) == 0)
 682         {
 683             _Bool tmp = 0;
 684             status = cf_util_get_boolean (child, &tmp);
 685             if (tmp)
 686                 conf->delivery_mode = CAMQP_DM_PERSISTENT;
 687             else
 688                 conf->delivery_mode = CAMQP_DM_VOLATILE;
 689         }
 690         else if ((strcasecmp ("StoreRates", child->key) == 0) && publish)
 691             status = cf_util_get_boolean (child, &conf->store_rates);
 692         else
 693             WARNING ("amqp plugin: Ignoring unknown "
 694                     "configuration option \"%s\".", child->key);
 695
 696         if (status != 0)
 697             break;
 698     } /* for (i = 0; i < ci->children_num; i++) */
 699
 700     if ((status == 0) && !publish && (conf->exchange == NULL))
 701     {
 702         if (conf->routingkey != NULL)
 703             WARNING ("amqp plugin: The option \"RoutingKey\" was given "
 704                     "without the \"Exchange\" option. It will be ignored.");
 705
 706         if (conf->exchange_type != NULL)
 707             WARNING ("amqp plugin: The option \"ExchangeType\" was given "
 708                     "without the \"Exchange\" option. It will be ignored.");
 709     }
 710
 711     if (status != 0)
 712     {
 713         camqp_config_free (conf);
 714         return (status);
 715     }
 716
 717     if (conf->exchange != NULL)
 718     {
 719         DEBUG ("amqp plugin: camqp_config_connection: exchange = %s;",
 720                 conf->exchange);
 721     }
 722
 723     if (publish)
 724     {
 725         char cbname[128];
 726         user_data_t ud = { conf, camqp_config_free };
 727
 728         ssnprintf (cbname, sizeof (cbname), "amqp/%s", conf->name);
 729
 730         status = plugin_register_write (cbname, camqp_write, &ud);
 731         if (status != 0)
 732         {
 733             camqp_config_free (conf);
 734             return (status);
 735         }
 736     }
 737     else
 738     {
 739         status = camqp_subscribe_init (conf);
 740         if (status != 0)
 741         {
 742             camqp_config_free (conf);
 743             return (status);
 744         }
 745     }
 746
 747     return (0);
 748 } /* }}} int camqp_config_connection */
 749
 750 static int camqp_config (oconfig_item_t *ci) /* {{{ */
 751 {
 752     int i;
 753
 754     for (i = 0; i < ci->children_num; i++)
 755     {
 756         oconfig_item_t *child = ci->children + i;
 757
 758         if (strcasecmp ("Publish", child->key) == 0)
 759             camqp_config_connection (child, /* publish = */ 1);
 760         else if (strcasecmp ("Subscribe", child->key) == 0)
 761             camqp_config_connection (child, /* publish = */ 0);
 762         else
 763             WARNING ("amqp plugin: Ignoring unknown config option \"%s\".",
 764                     child->key);
 765     } /* for (ci->children_num) */
 766
 767     return (0);
 768 } /* }}} int camqp_config */
 769
 770 static int shutdown (void) /* {{{ */
 771 {
 772     size_t i;
 773
 774     DEBUG ("amqp plugin: Shutting down %zu subscriber threads.",
 775             subscriber_threads_num);
 776
 777     subscriber_threads_running = 0;
 778     for (i = 0; i < subscriber_threads_num; i++)
 779     {
 780         /* FIXME: Sending a signal is not very elegant here. Maybe find out how
 781          * to use a timeout in the thread and check for the variable in regular
 782          * intervals. */
 783         pthread_kill (subscriber_threads[i], SIGTERM);
 784         pthread_join (subscriber_threads[i], /* retval = */ NULL);
 785     }
 786
 787     subscriber_threads_num = 0;
 788     sfree (subscriber_threads);
 789
 790     DEBUG ("amqp plugin: All subscriber threads exited.");
 791
 792     return (0);
 793 } /* }}} int shutdown */
 794
 795 void module_register (void)
 796 {
 797     plugin_register_complex_config ("amqp", camqp_config);
 798     plugin_register_shutdown ("amqp", shutdown);
 799 } /* void module_register */
 800
 801 /* vim: set sw=4 sts=4 et fdm=marker : */