X-Git-Url: https://git.octo.it/?a=blobdiff_plain;f=src%2Famqp.c;h=bebaea7c8e7eb4bdbbcff93684321a851fc94cbd;hb=1a477ecb462094ad9e13320a9234716ead038b9c;hp=c9929dc0b12e00a9a925ebc6c98c559162a27167;hpb=531f59523e969b498c570f9f7083bf41b6811773;p=collectd.git diff --git a/src/amqp.c b/src/amqp.c index c9929dc0..bebaea7c 100644 --- a/src/amqp.c +++ b/src/amqp.c @@ -38,6 +38,18 @@ #include #include +#ifdef HAVE_AMQP_TCP_SOCKET_H +# include +#endif +#ifdef HAVE_AMQP_SOCKET_H +# include +#endif +#if defined HAVE_DECL_AMQP_SOCKET_CLOSE && !HAVE_DECL_AMQP_SOCKET_CLOSE +/* rabbitmq-c does not currently ship amqp_socket.h + * and, thus, does not define this function. */ +int amqp_socket_close(amqp_socket_t *); +#endif + /* Defines for the delivery mode. I have no idea why they're not defined by the * library.. */ #define CAMQP_DM_VOLATILE 1 @@ -74,6 +86,7 @@ struct camqp_config_s char *prefix; char *postfix; char escape_char; + unsigned int graphite_flags; /* subscribe only */ char *exchange_type; @@ -389,8 +402,12 @@ static int camqp_setup_queue (camqp_config_t *conf) /* {{{ */ static int camqp_connect (camqp_config_t *conf) /* {{{ */ { amqp_rpc_reply_t reply; - int sockfd; int status; +#ifdef HAVE_AMQP_TCP_SOCKET + amqp_socket_t *socket; +#else + int sockfd; +#endif if (conf->connection != NULL) return (0); @@ -402,6 +419,34 @@ static int camqp_connect (camqp_config_t *conf) /* {{{ */ return (ENOMEM); } +#ifdef HAVE_AMQP_TCP_SOCKET +# define CLOSE_SOCKET() amqp_socket_close (socket) + /* TODO: add support for SSL using amqp_ssl_socket_new + * and related functions */ + socket = amqp_tcp_socket_new (conf->connection); + if (! socket) + { + ERROR ("amqp plugin: amqp_tcp_socket_new failed."); + amqp_destroy_connection (conf->connection); + conf->connection = NULL; + return (ENOMEM); + } + + status = amqp_socket_open (socket, CONF(conf, host), conf->port); + if (status < 0) + { + char errbuf[1024]; + status *= -1; + ERROR ("amqp plugin: amqp_socket_open failed: %s", + sstrerror (status, errbuf, sizeof (errbuf))); + CLOSE_SOCKET (); + amqp_destroy_connection (conf->connection); + conf->connection = NULL; + return (status); + } +#else /* HAVE_AMQP_TCP_SOCKET */ +# define CLOSE_SOCKET close(sockfd) + /* this interface is deprecated as of rabbitmq-c 0.4 */ sockfd = amqp_open_socket (CONF(conf, host), conf->port); if (sockfd < 0) { @@ -414,6 +459,7 @@ static int camqp_connect (camqp_config_t *conf) /* {{{ */ return (status); } amqp_set_sockfd (conf->connection, sockfd); +#endif reply = amqp_login (conf->connection, CONF(conf, vhost), /* channel max = */ 0, @@ -426,7 +472,7 @@ static int camqp_connect (camqp_config_t *conf) /* {{{ */ ERROR ("amqp plugin: amqp_login (vhost = %s, user = %s) failed.", CONF(conf, vhost), CONF(conf, user)); amqp_destroy_connection (conf->connection); - close (sockfd); + CLOSE_SOCKET (); conf->connection = NULL; return (1); } @@ -439,7 +485,7 @@ static int camqp_connect (camqp_config_t *conf) /* {{{ */ ERROR ("amqp plugin: amqp_channel_open failed."); amqp_connection_close (conf->connection, AMQP_REPLY_SUCCESS); amqp_destroy_connection (conf->connection); - close(sockfd); + CLOSE_SOCKET (); conf->connection = NULL; return (1); } @@ -600,6 +646,8 @@ static void *camqp_subscribe_thread (void *user_data) /* {{{ */ camqp_config_t *conf = user_data; int status; + cdtime_t interval = plugin_get_interval (); + while (subscriber_threads_running) { amqp_frame_t frame; @@ -610,8 +658,8 @@ static void *camqp_subscribe_thread (void *user_data) /* {{{ */ struct timespec ts_interval; ERROR ("amqp plugin: camqp_connect failed. " "Will sleep for %.3f seconds.", - CDTIME_T_TO_DOUBLE (interval_g)); - CDTIME_T_TO_TIMESPEC (interval_g, &ts_interval); + CDTIME_T_TO_DOUBLE (interval)); + CDTIME_T_TO_TIMESPEC (interval, &ts_interval); nanosleep (&ts_interval, /* remaining = */ NULL); continue; } @@ -622,9 +670,9 @@ static void *camqp_subscribe_thread (void *user_data) /* {{{ */ struct timespec ts_interval; ERROR ("amqp plugin: amqp_simple_wait_frame failed. " "Will sleep for %.3f seconds.", - CDTIME_T_TO_DOUBLE (interval_g)); + CDTIME_T_TO_DOUBLE (interval)); camqp_close_connection (conf); - CDTIME_T_TO_TIMESPEC (interval_g, &ts_interval); + CDTIME_T_TO_TIMESPEC (interval, &ts_interval); nanosleep (&ts_interval, /* remaining = */ NULL); continue; } @@ -650,6 +698,7 @@ static void *camqp_subscribe_thread (void *user_data) /* {{{ */ camqp_config_free (conf); pthread_exit (NULL); + return (NULL); } /* }}} void *camqp_subscribe_thread */ static int camqp_subscribe_init (camqp_config_t *conf) /* {{{ */ @@ -669,7 +718,7 @@ static int camqp_subscribe_init (camqp_config_t *conf) /* {{{ */ tmp = subscriber_threads + subscriber_threads_num; memset (tmp, 0, sizeof (*tmp)); - status = pthread_create (tmp, /* attr = */ NULL, + status = plugin_thread_create (tmp, /* attr = */ NULL, camqp_subscribe_thread, conf); if (status != 0) { @@ -790,7 +839,8 @@ static int camqp_write (const data_set_t *ds, const value_list_t *vl, /* {{{ */ else if (conf->format == CAMQP_FORMAT_GRAPHITE) { status = format_graphite (buffer, sizeof (buffer), ds, vl, - conf->prefix, conf->postfix, conf->escape_char); + conf->prefix, conf->postfix, conf->escape_char, + conf->graphite_flags); if (status != 0) { ERROR ("amqp plugin: format_graphite failed with status %i.", @@ -930,7 +980,11 @@ static int camqp_config_connection (oconfig_item_t *ci, /* {{{ */ conf->delivery_mode = CAMQP_DM_VOLATILE; } else if ((strcasecmp ("StoreRates", child->key) == 0) && publish) + { status = cf_util_get_boolean (child, &conf->store_rates); + (void) cf_util_get_flag (child, &conf->graphite_flags, + GRAPHITE_STORE_RATES); + } else if ((strcasecmp ("Format", child->key) == 0) && publish) status = camqp_config_set_format (child, conf); else if ((strcasecmp ("GraphitePrefix", child->key) == 0) && publish)