*
* Acquires the host lock, disconnects on errors.
*/
-static int wrr_send(struct riemann_host *host, riemann_message_t *msg) /* {{{ */
+static int wrr_send_nolock(struct riemann_host *host, riemann_message_t *msg) /* {{{ */
{
int status = 0;
- pthread_mutex_lock (&host->lock);
status = wrr_connect(host);
if (status != 0) {
- pthread_mutex_unlock(&host->lock);
return status;
}
status = riemann_client_send_message(host->client, msg);
if (status != 0) {
wrr_disconnect(host);
- pthread_mutex_unlock(&host->lock);
return status;
}
if (response == NULL)
{
wrr_disconnect(host);
- pthread_mutex_unlock(&host->lock);
return errno;
}
riemann_message_free(response);
}
- pthread_mutex_unlock (&host->lock);
return 0;
} /* }}} int wrr_send */
+static int wrr_send(struct riemann_host *host, riemann_message_t *msg)
+{
+ int status = 0;
+
+ pthread_mutex_lock (&host->lock);
+ status = wrr_send_nolock(host, msg);
+ pthread_mutex_unlock (&host->lock);
+ return status;
+}
+
static riemann_message_t *wrr_notification_to_message(struct riemann_host *host, /* {{{ */
notification_t const *n)
{
cdtime_t now;
int status = 0;
+ now = cdtime();
if (timeout > 0) {
- now = cdtime();
- if ((host->batch_init + timeout) > now)
+ if ((host->batch_init + timeout) > now) {
return status;
+ }
}
- wrr_send(host, host->batch_msg);
+ wrr_send_nolock(host, host->batch_msg);
riemann_message_free(host->batch_msg);
- if (host->client_type != RIEMANN_CLIENT_UDP)
- {
- riemann_message_t *response;
-
- response = riemann_client_recv_message(host->client);
-
- if (!response)
- {
- wrr_disconnect(host);
- return errno;
- }
-
- riemann_message_free(response);
- }
-
- host->batch_init = cdtime();
+ host->batch_init = now;
host->batch_msg = NULL;
return status;
}
}
if (host->client_type != RIEMANN_CLIENT_UDP && host->batch_mode) {
- wrr_batch_add_value_list(host, ds, vl, statuses);
+ wrr_batch_add_value_list(host, ds, vl, statuses);
} else {
msg = wrr_value_list_to_message(host, ds, vl, statuses);
if (msg == NULL)