4 ** Copyright (c) <2009> <sebastien.pahl@dotcloud.com>
6 ** Permission is hereby granted, free of charge, to any person
7 ** obtaining a copy of this software and associated documentation
8 ** files (the "Software"), to deal in the Software without
9 ** restriction, including without limitation the rights to use,
10 ** copy, modify, merge, publish, distribute, sublicense, and/or sell
11 ** copies of the Software, and to permit persons to whom the
12 ** Software is furnished to do so, subject to the following
15 ** The above copyright notice and this permission notice shall be
16 ** included in all copies or substantial portions of the Software.
18 ** THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
19 ** EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
20 ** OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
21 ** NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
22 ** HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
23 ** WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
24 ** FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
25 ** OTHER DEALINGS IN THE SOFTWARE.
38 #include "utils_format_json.h"
41 #include <amqp_framing.h>
43 /* Defines for the delivery mode. I have no idea why they're not defined by the
45 #define AMQP_DM_VOLATILE 1
46 #define AMQP_DM_PERSISTENT 2
49 static char *host = NULL;
50 static char *vhost = NULL;
51 static char *user = NULL;
52 static char *password = NULL;
53 static char *exchange = NULL;
54 static char *routingkey = NULL;
55 static uint8_t delivery_mode = AMQP_DM_VOLATILE;
57 static amqp_connection_state_t amqp_conn = NULL;
58 static pthread_mutex_t amqp_conn_lock = PTHREAD_MUTEX_INITIALIZER;
60 static const char *config_keys[] =
72 static int config_keys_num = STATIC_ARRAY_SIZE(config_keys);
74 static int config_set(char **var, const char *value)
77 if ((*var = strdup(value)) == NULL)
82 static int config(const char *key, const char *value)
84 if (strcasecmp(key, "host") == 0)
85 return (config_set(&host, value));
86 else if(strcasecmp(key, "port") == 0)
90 tmp = service_name_to_port_number (value);
93 ERROR ("AMQP plugin: Cannot parse `%s' as a "
94 "service name (port number).", value);
101 else if (strcasecmp(key, "vhost") == 0)
102 return (config_set(&vhost, value));
103 else if (strcasecmp(key, "user") == 0)
104 return (config_set(&user, value));
105 else if (strcasecmp(key, "password") == 0)
106 return (config_set(&password, value));
107 else if (strcasecmp(key, "exchange") == 0)
108 return (config_set(&exchange, value));
109 else if (strcasecmp(key, "routingkey") == 0)
110 return (config_set(&routingkey, value));
111 else if (strcasecmp ("Persistent", key) == 0)
114 delivery_mode = AMQP_DM_PERSISTENT;
116 delivery_mode = AMQP_DM_VOLATILE;
122 static int amqp_write_locked (const char *buffer)
124 amqp_rpc_reply_t reply;
125 amqp_basic_properties_t props;
128 if (amqp_conn == NULL)
132 amqp_conn = amqp_new_connection ();
133 if (amqp_conn == NULL)
135 ERROR ("amqp plugin: amqp_new_connection failed.");
139 sockfd = amqp_open_socket (host, port);
143 status = (-1) * sockfd;
144 ERROR ("amqp plugin: amqp_open_socket failed: %s",
145 sstrerror (status, errbuf, sizeof (errbuf)));
146 amqp_destroy_connection(amqp_conn);
151 amqp_set_sockfd (amqp_conn, sockfd);
153 reply = amqp_login(amqp_conn, vhost,
154 /* channel max = */ 0,
155 /* frame max = */ 131072,
157 /* authentication: */ AMQP_SASL_METHOD_PLAIN, user, password);
158 if (reply.reply_type != AMQP_RESPONSE_NORMAL)
160 ERROR ("amqp plugin: amqp_login (vhost = %s, user = %s) failed.",
162 amqp_destroy_connection(amqp_conn);
168 amqp_channel_open (amqp_conn, /* channel = */ 1);
169 /* FIXME: Is checking "reply.reply_type" really correct here? How does
170 * it get set? --octo */
171 if (reply.reply_type != AMQP_RESPONSE_NORMAL)
173 ERROR ("amqp plugin: amqp_channel_open failed.");
174 amqp_connection_close (amqp_conn, AMQP_REPLY_SUCCESS);
175 amqp_destroy_connection(amqp_conn);
181 INFO ("amqp plugin: Successfully opened connection to vhost \"%s\" "
182 "on %s:%i.", vhost, host, port);
183 } /* if (amqp_conn == NULL) */
185 props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
186 props.content_type = amqp_cstring_bytes("application/json");
187 props.delivery_mode = delivery_mode;
189 status = amqp_basic_publish(amqp_conn,
191 amqp_cstring_bytes(exchange),
192 amqp_cstring_bytes(routingkey),
196 amqp_cstring_bytes(buffer));
201 ERROR ("amqp plugin: amqp_basic_publish failed with status %i.",
204 sockfd = amqp_get_sockfd (amqp_conn);
205 amqp_channel_close (amqp_conn, 1, AMQP_REPLY_SUCCESS);
206 amqp_connection_close (amqp_conn, AMQP_REPLY_SUCCESS);
207 amqp_destroy_connection (amqp_conn);
213 } /* int amqp_write_locked */
215 static int amqp_write (const data_set_t *ds, const value_list_t *vl,
216 __attribute__((unused)) user_data_t *user_data)
223 if ((ds == NULL) || (vl == NULL))
226 memset (buffer, 0, sizeof (buffer));
227 bfree = sizeof (buffer);
230 format_json_initialize(buffer, &bfill, &bfree);
231 /* TODO: Possibly add a config option "StoreRates" and pass the value along here. */
232 format_json_value_list(buffer, &bfill, &bfree, ds, vl, /* rates = */ 0);
233 format_json_finalize(buffer, &bfill, &bfree);
235 pthread_mutex_lock (&amqp_conn_lock);
236 status = amqp_write_locked (buffer);
237 pthread_mutex_unlock (&amqp_conn_lock);
240 } /* int amqp_write */
242 static int shutdown(void)
244 pthread_mutex_lock (&amqp_conn_lock);
245 if (amqp_conn != NULL)
249 sockfd = amqp_get_sockfd (amqp_conn);
250 amqp_channel_close (amqp_conn, 1, AMQP_REPLY_SUCCESS);
251 amqp_connection_close (amqp_conn, AMQP_REPLY_SUCCESS);
252 amqp_destroy_connection (amqp_conn);
256 pthread_mutex_unlock (&amqp_conn_lock);
268 void module_register(void)
270 plugin_register_config("amqp", config, config_keys, config_keys_num);
271 plugin_register_write("amqp", amqp_write, NULL);
272 plugin_register_shutdown("amqp", shutdown);
275 /* vim: set sw=4 sts=4 et : */