diff --git a/src/amqp.c b/src/amqp.c
index c77c345bbcd071b1a5fc9a8c150e8d5957b47cfc..89284c81962ea127d4c901ee0c64e7aceb13275a 100644 (file)
--- a/src/amqp.c
+++ b/src/amqp.c
/**
* collectd - src/amqp.c
- * Copyright (C) 2009 Sebastien Pahl
- * Copyright (C) 2010 Florian Forster
+ * Copyright (C) 2009 Sebastien Pahl
+ * Copyright (C) 2010-2012 Florian Forster
*
* Permission is hereby granted, free of charge, to any person obtaining a
* copy of this software and associated documentation files (the "Software"),
* Florian Forster <octo at verplant.org>
**/
-#include <stdint.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <strings.h>
-#include <pthread.h>
-
#include "collectd.h"
#include "common.h"
#include "plugin.h"
#include "utils_cmd_putval.h"
#include "utils_format_json.h"
+#include <pthread.h>
+
#include <amqp.h>
#include <amqp_framing.h>
break;
case AMQP_RESPONSE_LIBRARY_EXCEPTION:
+#if HAVE_AMQP_RPC_REPLY_T_LIBRARY_ERRNO
if (r.library_errno)
return (sstrerror (r.library_errno, buffer, buffer_size));
+#else
+ if (r.library_error)
+ return (sstrerror (r.library_error, buffer, buffer_size));
+#endif
else
sstrncpy (buffer, "End of stream", sizeof (buffer));
break;
return (buffer);
} /* }}} char *camqp_strerror */
+#if HAVE_AMQP_RPC_REPLY_T_LIBRARY_ERRNO
static int camqp_create_exchange (camqp_config_t *conf) /* {{{ */
{
amqp_exchange_declare_ok_t *ed_ret;
return (0);
} /* }}} int camqp_create_exchange */
+#else
+static int camqp_create_exchange (camqp_config_t *conf) /* {{{ */
+{
+ amqp_exchange_declare_ok_t *ed_ret;
+ amqp_table_t argument_table;
+ struct amqp_table_entry_t_ argument_table_entries[1];
+
+ if (conf->exchange_type == NULL)
+ return (0);
+
+ /* Valid arguments: "auto_delete", "internal" */
+ argument_table.num_entries = STATIC_ARRAY_SIZE (argument_table_entries);
+ argument_table.entries = argument_table_entries;
+ argument_table_entries[0].key = amqp_cstring_bytes ("auto_delete");
+ argument_table_entries[0].value.kind = AMQP_FIELD_KIND_BOOLEAN;
+ argument_table_entries[0].value.value.boolean = 1;
+
+ ed_ret = amqp_exchange_declare (conf->connection,
+ /* channel = */ CAMQP_CHANNEL,
+ /* exchange = */ amqp_cstring_bytes (conf->exchange),
+ /* type = */ amqp_cstring_bytes (conf->exchange_type),
+ /* passive = */ 0,
+ /* durable = */ 0,
+ /* arguments = */ argument_table);
+ if ((ed_ret == NULL) && camqp_is_error (conf))
+ {
+ char errbuf[1024];
+ ERROR ("amqp plugin: amqp_exchange_declare failed: %s",
+ camqp_strerror (conf, errbuf, sizeof (errbuf)));
+ camqp_close_connection (conf);
+ return (-1);
+ }
+
+ INFO ("amqp plugin: Successfully created exchange \"%s\" "
+ "with type \"%s\".",
+ conf->exchange, conf->exchange_type);
+
+ return (0);
+} /* }}} int camqp_create_exchange */
+#endif
static int camqp_setup_queue (camqp_config_t *conf) /* {{{ */
{
/* consumer_tag = */ AMQP_EMPTY_BYTES,
/* no_local = */ 0,
/* no_ack = */ 1,
- /* exclusive = */ 0);
+ /* exclusive = */ 0,
+ /* arguments = */ AMQP_EMPTY_TABLE
+ );
if ((cm_ret == NULL) && camqp_is_error (conf))
{
char errbuf[1024];
return (0);
} /* }}} int camqp_connect */
-static int shutdown (void) /* {{{ */
+static int camqp_shutdown (void) /* {{{ */
{
size_t i;
DEBUG ("amqp plugin: All subscriber threads exited.");
return (0);
-} /* }}} int shutdown */
+} /* }}} int camqp_shutdown */
/*
* Subscribing code
status = camqp_connect (conf);
if (status != 0)
{
+ struct timespec ts_interval;
ERROR ("amqp plugin: camqp_connect failed. "
- "Will sleep for %i seconds.", interval_g);
- sleep (interval_g);
+ "Will sleep for %.3f seconds.",
+ CDTIME_T_TO_DOUBLE (interval_g));
+ CDTIME_T_TO_TIMESPEC (interval_g, &ts_interval);
+ nanosleep (&ts_interval, /* remaining = */ NULL);
continue;
}
status = amqp_simple_wait_frame (conf->connection, &frame);
if (status < 0)
{
+ struct timespec ts_interval;
ERROR ("amqp plugin: amqp_simple_wait_frame failed. "
- "Will sleep for %i seconds.", interval_g);
+ "Will sleep for %.3f seconds.",
+ CDTIME_T_TO_DOUBLE (interval_g));
camqp_close_connection (conf);
- sleep (interval_g);
+ CDTIME_T_TO_TIMESPEC (interval_g, &ts_interval);
+ nanosleep (&ts_interval, /* remaining = */ NULL);
continue;
}
camqp_config_free (conf);
pthread_exit (NULL);
+ return (NULL);
} /* }}} void *camqp_subscribe_thread */
static int camqp_subscribe_init (camqp_config_t *conf) /* {{{ */
/*
* Publishing code
*/
+/* XXX: You must hold "conf->lock" when calling this function! */
static int camqp_write_locked (camqp_config_t *conf, /* {{{ */
const char *buffer, const char *routing_key)
{
void module_register (void)
{
plugin_register_complex_config ("amqp", camqp_config);
- plugin_register_shutdown ("amqp", shutdown);
+ plugin_register_shutdown ("amqp", camqp_shutdown);
} /* void module_register */
/* vim: set sw=4 sts=4 et fdm=marker : */