};
typedef struct ce_connection_info_s ce_connection_info_t;
+struct ce_callback_info_s
+{
+ int fd;
+ ETERM *fun;
+};
+typedef struct ce_callback_info_s ce_callback_info_t;
+
/*
* Private variables
*/
static char conf_nodename[256] = "collectd";
static char conf_fullname[256] = "collectd@alyja.office.noris.de";
+static int connection_counter = 1;
+static pthread_mutex_t connection_lock = PTHREAD_MUTEX_INITIALIZER;
+
/*
* Private functions
*/
+static void ce_free_callback_info (ce_callback_info_t *ci) /* {{{ */
+{
+ if (ci == NULL)
+ return;
+
+ if (ci->fd >= 0)
+ {
+ erl_close_connection (ci->fd);
+ ci->fd = -1;
+ }
+
+ if (ci->fun != NULL)
+ erl_free_compound (ci->fun);
+
+ free (ci);
+} /* }}} void ce_free_callback_info */
+
static int send_atom (int fd, ETERM *to, const char *atom) /* {{{ */
{
ETERM *reply;
else
status = 0;
- erl_free_term (reply);
+ erl_free_compound (reply);
return (status);
} /* }}} int send_error */
return (0);
} /* }}} int eterm_to_value_list */
+static int ce_read (user_data_t *ud) /* {{{ */
+{
+ ce_callback_info_t *ci;
+ ETERM *rpc_args;
+ ETERM *rpc_reply;
+
+ if ((ud == NULL) || (ud->data == NULL))
+ return (-1);
+
+ ci = ud->data;
+
+ rpc_args = erl_format ("[~w,[]]", erl_copy_term (ci->fun));
+ if (rpc_args == NULL)
+ {
+ ERROR ("erlang plugin: erl_format failed.");
+ return (-1);
+ }
+
+ DEBUG ("erlang plugin: Making remote procedure call ...");
+ rpc_reply = erl_rpc (ci->fd,
+ /* module = */ "erlang", /* function = */ "apply",
+ /* arguments = */ rpc_args);
+ DEBUG ("erlang plugin: ... done.");
+ erl_free_compound (rpc_args);
+ if (rpc_reply == NULL)
+ {
+ char errbuf[1024];
+ ERROR ("erlang plugin: erl_rpc failed: %s",
+ sstrerror (erl_errno, errbuf, sizeof (errbuf)));
+ return (-1);
+ }
+
+ /* FIXME: The return value is not yet used. */
+ erl_free_compound (rpc_reply);
+
+ return (0);
+} /* }}} int ce_read */
+
/* Returns non-zero only if the request could not be handled gracefully. */
static int handle_dispatch_values (ce_connection_info_t *cinfo, /* {{{ */
const ErlMessage *req)
if (status != 0)
{
free (vl.values);
- send_error (cinfo->fd, req->from, "Cannot parse argument as value list.");
- return (0);
+ status = send_error (cinfo->fd, req->from, "Cannot parse argument as value list.");
+ return (status);
}
status = plugin_dispatch_values (&vl);
if (status != 0)
{
free (vl.values);
- send_error (cinfo->fd, req->from, "plugin_dispatch_values failed.");
- return (0);
+ status = send_error (cinfo->fd, req->from, "plugin_dispatch_values failed.");
+ return (status);
}
free (vl.values);
- send_atom (cinfo->fd, req->from, "success");
+ status = send_atom (cinfo->fd, req->from, "success");
- return (0);
+ return (status);
+} /* }}} int handle_dispatch_values */
+
+/* Returns non-zero only if the request could not be handled gracefully. */
+static int handle_register_read (ce_connection_info_t *cinfo, /* {{{ */
+ const ErlMessage *req)
+{
+ ETERM *eterm_cb;
+ ce_callback_info_t *ci;
+ user_data_t ud;
+ int status;
+ int connection_number;
+ char callback_name[64];
+
+ if ((cinfo == NULL) || (req == NULL))
+ return (EINVAL);
+
+ eterm_cb = erl_element (2, req->msg);
+
+ if (ERL_TYPE (eterm_cb) != ERL_FUNCTION)
+ {
+ erl_free_term (eterm_cb);
+ status = send_error (cinfo->fd, req->from,
+ "Argument to `register_read' must be a callback function.");
+ return (status);
+ }
+
+ ci = malloc (sizeof (ci));
+ if (ci == NULL)
+ {
+ erl_free_term (eterm_cb);
+ status = send_error (cinfo->fd, req->from, "malloc failed.");
+ return (status);
+ }
+
+ /* Lock around `erl_connect_init' and `erl_connect'. */
+ pthread_mutex_lock (&connection_lock);
+
+ connection_number = connection_counter;
+ connection_counter++;
+
+ /* Create a new `cnode' for each connection. Otherwise we cannot determine
+ * which RPC call a message belongs to. */
+ status = erl_connect_init (connection_number, conf_cookie,
+ /* creation = */ 0);
+ if (!status) /* Yes, it's this way around in this case ... {{{ */
+ {
+ char errbuf[1024];
+ pthread_mutex_unlock (&connection_lock);
+ ERROR ("erlang plugin: erl_connect_init failed: %s",
+ sstrerror (erl_errno, errbuf, sizeof (errbuf)));
+ sfree (ci);
+ erl_free_term (eterm_cb);
+ status = send_error (cinfo->fd, req->from, "erl_connect failed.");
+ return (status);
+ } /* }}} */
+
+ ci->fd = erl_connect (cinfo->conn.nodename);
+ if (ci->fd < 0) /* {{{ */
+ {
+ char errbuf[1024];
+ pthread_mutex_unlock (&connection_lock);
+ ERROR ("erlang plugin: erl_connect(%s) failed: %s",
+ cinfo->conn.nodename,
+ sstrerror (erl_errno, errbuf, sizeof (errbuf)));
+ sfree (ci);
+ erl_free_term (eterm_cb);
+ status = send_error (cinfo->fd, req->from, "erl_connect failed.");
+ return (status);
+ } /* }}} */
+
+ pthread_mutex_unlock (&connection_lock);
+
+ ci->fun = eterm_cb;
+
+ memset (&ud, 0, sizeof (ud));
+ ud.data = ci;
+ ud.free_func = (void (*) (void *)) ce_free_callback_info;
+
+ ssnprintf (callback_name, sizeof (callback_name), "erlang:%i",
+ connection_number);
+
+ status = plugin_register_complex_read (callback_name,
+ ce_read, /* interval = */ NULL, &ud);
+ if (status == 0)
+ status = send_atom (cinfo->fd, req->from, "success");
+ else
+ status = send_error (cinfo->fd, req->from,
+ "plugin_register_complex_read failed.");
+
+ return (status);
} /* }}} int handle_dispatch_values */
static void *handle_client_thread (void *arg) /* {{{ */
reply = NULL;
if (strcmp ("dispatch_values", ERL_ATOM_PTR (func)) == 0)
status = handle_dispatch_values (cinfo, &emsg);
+ else if (strcmp ("register_read", ERL_ATOM_PTR (func)) == 0)
+ status = handle_register_read (cinfo, &emsg);
else
{
ERROR ("erlang plugin: Received request for invalid function `%s'.",
fd = erl_accept (listen, &conn);
if (fd < 0)
{
- ERROR ("erlang plugin: erl_accept failed with status %i.", fd);
- close (listen);
- exit (EXIT_FAILURE);
+ char errbuf[1024];
+ ERROR ("erlang plugin: erl_accept failed: %s",
+ sstrerror (erl_errno, errbuf, sizeof (errbuf)));
+ continue;
}
DEBUG ("erlang plugin: Got connection from %s on fd %i.",
conn.nodename, fd);
projects / collectd.git / blobdiff