X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Famqp.c;h=ec7943022b8db23177dae9e284e5a5fe6717abcf;hb=d7ddea73a881d194933bbfbfd0e13efff8ac9296;hp=1764129faf4a57fdc622062156d28ffb322a4bef;hpb=ca316d91e178412604ea8462dc60a8bc32cbfc87;p=collectd.git

diff --git a/src/amqp.c b/src/amqp.c
index 1764129f..ec794302 100644
--- a/src/amqp.c
+++ b/src/amqp.c
@@ -80,6 +80,9 @@ struct camqp_config_s
     char   *exchange;
     char   *routing_key;
 
+    /* Number of seconds to wait before connection is retried */
+    int     connection_retry_delay;
+
     /* publish only */
     uint8_t delivery_mode;
     _Bool   store_rates;
@@ -196,11 +199,11 @@ static char *camqp_strerror (camqp_config_t *conf, /* {{{ */
     switch (r.reply_type)
     {
         case AMQP_RESPONSE_NORMAL:
-            sstrncpy (buffer, "Success", sizeof (buffer));
+            sstrncpy (buffer, "Success", buffer_size);
             break;
 
         case AMQP_RESPONSE_NONE:
-            sstrncpy (buffer, "Missing RPC reply type", sizeof (buffer));
+            sstrncpy (buffer, "Missing RPC reply type", buffer_size);
             break;
 
         case AMQP_RESPONSE_LIBRARY_EXCEPTION:
@@ -212,7 +215,7 @@ static char *camqp_strerror (camqp_config_t *conf, /* {{{ */
                 return (sstrerror (r.library_error, buffer, buffer_size));
 #endif
             else
-                sstrncpy (buffer, "End of stream", sizeof (buffer));
+                sstrncpy (buffer, "End of stream", buffer_size);
             break;
 
         case AMQP_RESPONSE_SERVER_EXCEPTION:
@@ -301,6 +304,10 @@ static int camqp_create_exchange (camqp_config_t *conf) /* {{{ */
             /* type        = */ amqp_cstring_bytes (conf->exchange_type),
             /* passive     = */ 0,
             /* durable     = */ 0,
+#if defined(AMQP_VERSION) && AMQP_VERSION >= 0x00060000
+            /* auto delete = */ 0,
+            /* internal    = */ 0,
+#endif
             /* arguments   = */ argument_table);
     if ((ed_ret == NULL) && camqp_is_error (conf))
     {
@@ -405,6 +412,8 @@ static int camqp_setup_queue (camqp_config_t *conf) /* {{{ */
 
 static int camqp_connect (camqp_config_t *conf) /* {{{ */
 {
+    static time_t last_connect_time = 0;
+
     amqp_rpc_reply_t reply;
     int status;
 #ifdef HAVE_AMQP_TCP_SOCKET
@@ -416,6 +425,19 @@ static int camqp_connect (camqp_config_t *conf) /* {{{ */
     if (conf->connection != NULL)
         return (0);
 
+    time_t now = time(NULL);
+    if (now < (last_connect_time + conf->connection_retry_delay))
+    {
+        DEBUG("amqp plugin: skipping connection retry, "
+            "ConnectionRetryDelay: %d", conf->connection_retry_delay);
+        return(1);
+    }
+    else
+    {
+        DEBUG ("amqp plugin: retrying connection");
+        last_connect_time = now;
+    }
+
     conf->connection = amqp_new_connection ();
     if (conf->connection == NULL)
     {
@@ -694,7 +716,7 @@ static void *camqp_subscribe_thread (void *user_data) /* {{{ */
             continue;
         }
 
-        status = camqp_read_header (conf);
+        camqp_read_header (conf);
 
         amqp_maybe_release_buffers (conf->connection);
     } /* while (subscriber_threads_running) */
@@ -903,15 +925,14 @@ static int camqp_config_connection (oconfig_item_t *ci, /* {{{ */
     int status;
     int i;
 
-    conf = malloc (sizeof (*conf));
+    conf = calloc (1, sizeof (*conf));
     if (conf == NULL)
     {
-        ERROR ("amqp plugin: malloc failed.");
+        ERROR ("amqp plugin: calloc failed.");
         return (ENOMEM);
     }
 
     /* Initialize "conf" {{{ */
-    memset (conf, 0, sizeof (*conf));
     conf->publish = publish;
     conf->name = NULL;
     conf->format = CAMQP_FORMAT_COMMAND;
@@ -922,6 +943,8 @@ static int camqp_config_connection (oconfig_item_t *ci, /* {{{ */
     conf->password = NULL;
     conf->exchange = NULL;
     conf->routing_key = NULL;
+    conf->connection_retry_delay = 0;
+
     /* publish only */
     conf->delivery_mode = CAMQP_DM_VOLATILE;
     conf->store_rates = 0;
@@ -1017,6 +1040,8 @@ static int camqp_config_connection (oconfig_item_t *ci, /* {{{ */
             conf->escape_char = tmp_buff[0];
             sfree (tmp_buff);
         }
+        else if (strcasecmp ("ConnectionRetryDelay", child->key) == 0)
+            status = cf_util_get_int (child, &conf->connection_retry_delay);
         else
             WARNING ("amqp plugin: Ignoring unknown "
                     "configuration option \"%s\".", child->key);