+/*
+ * Subscribing code
+ */
+static int camqp_read_body (camqp_config_t *conf, /* {{{ */
+ size_t body_size, const char *content_type)
+{
+ char body[body_size + 1];
+ char *body_ptr;
+ size_t received;
+ amqp_frame_t frame;
+ int status;
+
+ memset (body, 0, sizeof (body));
+ body_ptr = &body[0];
+ received = 0;
+
+ while (received < body_size)
+ {
+ status = amqp_simple_wait_frame (conf->connection, &frame);
+ if (status < 0)
+ {
+ char errbuf[1024];
+ status = (-1) * status;
+ ERROR ("amqp plugin: amqp_simple_wait_frame failed: %s",
+ sstrerror (status, errbuf, sizeof (errbuf)));
+ camqp_close_connection (conf);
+ return (status);
+ }
+
+ if (frame.frame_type != AMQP_FRAME_BODY)
+ {
+ NOTICE ("amqp plugin: Unexpected frame type: %#"PRIx8,
+ frame.frame_type);
+ return (-1);
+ }
+
+ if ((body_size - received) < frame.payload.body_fragment.len)
+ {
+ WARNING ("amqp plugin: Body is larger than indicated by header.");
+ return (-1);
+ }
+
+ memcpy (body_ptr, frame.payload.body_fragment.bytes,
+ frame.payload.body_fragment.len);
+ body_ptr += frame.payload.body_fragment.len;
+ received += frame.payload.body_fragment.len;
+ } /* while (received < body_size) */
+
+ if (strcasecmp ("text/collectd", content_type) == 0)
+ {
+ status = handle_putval (stderr, body);
+ if (status != 0)
+ ERROR ("amqp plugin: handle_putval failed with status %i.",
+ status);
+ return (status);
+ }
+ else if (strcasecmp ("application/json", content_type) == 0)
+ {
+ ERROR ("amqp plugin: camqp_read_body: Parsing JSON data has not "
+ "been implemented yet. FIXME!");
+ return (0);
+ }
+ else
+ {
+ ERROR ("amqp plugin: camqp_read_body: Unknown content type \"%s\".",
+ content_type);
+ return (EINVAL);
+ }
+
+ /* not reached */
+ return (0);
+} /* }}} int camqp_read_body */
+
+static int camqp_read_header (camqp_config_t *conf) /* {{{ */
+{
+ int status;
+ amqp_frame_t frame;
+ amqp_basic_properties_t *properties;
+ char *content_type;
+
+ status = amqp_simple_wait_frame (conf->connection, &frame);
+ if (status < 0)
+ {
+ char errbuf[1024];
+ status = (-1) * status;
+ ERROR ("amqp plugin: amqp_simple_wait_frame failed: %s",
+ sstrerror (status, errbuf, sizeof (errbuf)));
+ camqp_close_connection (conf);
+ return (status);
+ }
+
+ if (frame.frame_type != AMQP_FRAME_HEADER)
+ {
+ NOTICE ("amqp plugin: Unexpected frame type: %#"PRIx8,
+ frame.frame_type);
+ return (-1);
+ }
+
+ properties = frame.payload.properties.decoded;
+ content_type = camqp_bytes_cstring (&properties->content_type);
+ if (content_type == NULL)
+ {
+ ERROR ("amqp plugin: Unable to determine content type.");
+ return (-1);
+ }
+
+ status = camqp_read_body (conf,
+ (size_t) frame.payload.properties.body_size,
+ content_type);
+
+ sfree (content_type);
+ return (status);
+} /* }}} int camqp_read_header */
+
+static void *camqp_subscribe_thread (void *user_data) /* {{{ */
+{
+ camqp_config_t *conf = user_data;
+ int status;
+
+ while (subscriber_threads_running)
+ {
+ amqp_frame_t frame;
+
+ status = camqp_connect (conf);
+ if (status != 0)
+ {
+ ERROR ("amqp plugin: camqp_connect failed. "
+ "Will sleep for %i seconds.", interval_g);
+ sleep (interval_g);
+ continue;
+ }
+
+ status = amqp_simple_wait_frame (conf->connection, &frame);
+ if (status < 0)
+ {
+ ERROR ("amqp plugin: amqp_simple_wait_frame failed. "
+ "Will sleep for %i seconds.", interval_g);
+ camqp_close_connection (conf);
+ sleep (interval_g);
+ continue;
+ }
+
+ if (frame.frame_type != AMQP_FRAME_METHOD)
+ {
+ DEBUG ("amqp plugin: Unexpected frame type: %#"PRIx8,
+ frame.frame_type);
+ continue;
+ }
+
+ if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
+ {
+ DEBUG ("amqp plugin: Unexpected method id: %#"PRIx32,
+ frame.payload.method.id);
+ continue;
+ }
+
+ status = camqp_read_header (conf);
+
+ amqp_maybe_release_buffers (conf->connection);
+ } /* while (subscriber_threads_running) */
+
+ camqp_config_free (conf);
+ pthread_exit (NULL);
+} /* }}} void *camqp_subscribe_thread */
+
+static int camqp_subscribe_init (camqp_config_t *conf) /* {{{ */
+{
+ int status;
+ pthread_t *tmp;
+
+ tmp = realloc (subscriber_threads,
+ sizeof (*subscriber_threads) * (subscriber_threads_num + 1));
+ if (tmp == NULL)
+ {
+ ERROR ("amqp plugin: realloc failed.");
+ camqp_config_free (conf);
+ return (ENOMEM);
+ }
+ subscriber_threads = tmp;
+ tmp = subscriber_threads + subscriber_threads_num;
+ memset (tmp, 0, sizeof (*tmp));
+
+ status = pthread_create (tmp, /* attr = */ NULL,
+ camqp_subscribe_thread, conf);
+ if (status != 0)
+ {
+ char errbuf[1024];
+ ERROR ("amqp plugin: pthread_create failed: %s",
+ sstrerror (status, errbuf, sizeof (errbuf)));
+ camqp_config_free (conf);
+ return (status);
+ }
+
+ subscriber_threads_num++;
+
+ return (0);
+} /* }}} int camqp_subscribe_init */
+
+/*
+ * 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)