1
/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
4
* Copyright (C) 2010 Marcus Eriksson
8
* Marcus Eriksson <krummas@gmail.com>
10
* This program is free software; you can redistribute it and/or modify
11
* it under the terms of the GNU General Public License as published by
12
* the Free Software Foundation; either version 2 of the License, or
13
* (at your option) any later version.
15
* This program is distributed in the hope that it will be useful,
16
* but WITHOUT ANY WARRANTY; without even the implied warranty of
17
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18
* GNU General Public License for more details.
20
* You should have received a copy of the GNU General Public License
21
* along with this program; if not, write to the Free Software
22
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
26
#include "rabbitmq_log.h"
27
#include <drizzled/message/transaction.pb.h>
28
#include <google/protobuf/io/coded_stream.h>
30
#include <drizzled/plugin/registry.h>
31
#include <drizzled/plugin.h>
33
#include "rabbitmq_handler.h"
36
using namespace drizzled;
37
using namespace google;
40
* The hostname to connect to
42
static char* sysvar_rabbitmq_host= NULL;
47
static int sysvar_rabbitmq_port= 0;
52
static char* sysvar_rabbitmq_username= NULL;
57
static char* sysvar_rabbitmq_password= NULL;
60
* rabbitmq virtualhost
62
static char* sysvar_rabbitmq_virtualhost= NULL;
65
* rabbitmq exchangename
67
static char* sysvar_rabbitmq_exchange= NULL;
70
* rabbitmq routing key
72
static char* sysvar_rabbitmq_routingkey= NULL;
75
* Is the rabbitmq log enabled?
77
static bool sysvar_rabbitmq_log_enabled= false;
80
* The name of the replicator plugin
81
* to pair the rabbitmq log's applier with.
82
* Defaults to "default"
84
static char *sysvar_rabbitmq_use_replicator= NULL;
85
static const char DEFAULT_USE_REPLICATOR[]= "default";
88
RabbitMQLog::RabbitMQLog(const string name_arg,
89
RabbitMQHandler* mqHandler)
90
:plugin::TransactionApplier(name_arg)
92
rabbitMQHandler= mqHandler;
95
RabbitMQLog::~RabbitMQLog()
99
plugin::ReplicationReturnCode
100
RabbitMQLog::apply(Session &, const message::Transaction &to_apply)
102
size_t message_byte_length= to_apply.ByteSize();
103
uint8_t* buffer= static_cast<uint8_t *>(malloc(message_byte_length));
106
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
108
return plugin::UNKNOWN_ERROR;
111
to_apply.SerializeWithCachedSizesToArray(buffer);
114
rabbitMQHandler->publish(buffer,
115
int(message_byte_length),
116
sysvar_rabbitmq_exchange,
117
sysvar_rabbitmq_routingkey);
121
errmsg_printf(ERRMSG_LVL_ERROR, _(e.what()));
123
return plugin::UNKNOWN_ERROR;
126
return plugin::SUCCESS;
129
static RabbitMQLog *rabbitmqLogger; ///< the actual plugin
130
static RabbitMQHandler* rabbitmqHandler; ///< the rabbitmq handler
133
* Initialize the rabbitmq logger - instanciates the dependencies (the handler)
134
* and creates the log handler with the dependency - makes it easier to swap out
135
* handler implementation
137
static int init(drizzled::plugin::Context &context)
139
if(sysvar_rabbitmq_log_enabled)
143
rabbitmqHandler= new RabbitMQHandler(sysvar_rabbitmq_host,
144
sysvar_rabbitmq_port,
145
sysvar_rabbitmq_username,
146
sysvar_rabbitmq_password,
147
sysvar_rabbitmq_virtualhost);
151
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler. Got error: %s\n"),
157
rabbitmqLogger= new RabbitMQLog("rabbit_log_applier", rabbitmqHandler);
161
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance. Got error: %s\n"),
166
context.add(rabbitmqLogger);
167
ReplicationServices &replication_services= ReplicationServices::singleton();
168
string replicator_name(sysvar_rabbitmq_use_replicator);
169
replication_services.attachApplier(rabbitmqLogger, replicator_name);
176
static DRIZZLE_SYSVAR_BOOL(enable,
177
sysvar_rabbitmq_log_enabled,
179
N_("Enable rabbitmq log"),
180
NULL, /* check func */
181
NULL, /* update func */
182
false /* default */);
185
static DRIZZLE_SYSVAR_STR(hostname,
186
sysvar_rabbitmq_host,
188
N_("Host name to connect to"),
189
NULL, /* check func */
190
NULL, /* update func*/
191
"localhost" /* default */);
194
static DRIZZLE_SYSVAR_INT(port,
195
sysvar_rabbitmq_port,
198
NULL, /* check func */
199
NULL, /* update func */
205
static DRIZZLE_SYSVAR_STR(username,
206
sysvar_rabbitmq_username,
208
N_("RabbitMQ username"),
209
NULL, /* check func */
210
NULL, /* update func*/
211
"guest" /* default */);
213
static DRIZZLE_SYSVAR_STR(password,
214
sysvar_rabbitmq_password,
216
N_("RabbitMQ password"),
217
NULL, /* check func */
218
NULL, /* update func*/
219
"guest" /* default */);
221
static DRIZZLE_SYSVAR_STR(virtualhost,
222
sysvar_rabbitmq_virtualhost,
224
N_("RabbitMQ virtualhost"),
225
NULL, /* check func */
226
NULL, /* update func*/
229
static DRIZZLE_SYSVAR_STR(exchange,
230
sysvar_rabbitmq_exchange,
232
N_("Name of RabbitMQ exchange to publish to"),
233
NULL, /* check func */
234
NULL, /* update func*/
235
"ReplicationExchange" /* default */);
237
static DRIZZLE_SYSVAR_STR(routingkey,
238
sysvar_rabbitmq_routingkey,
240
N_("Name of RabbitMQ routing key to use"),
241
NULL, /* check func */
242
NULL, /* update func*/
243
"ReplicationRoutingKey" /* default */);
245
static DRIZZLE_SYSVAR_STR(use_replicator,
246
sysvar_rabbitmq_use_replicator,
248
N_("Name of the replicator plugin to use (default='default_replicator')"),
249
NULL, /* check func */
250
NULL, /* update func*/
251
DEFAULT_USE_REPLICATOR /* default */);
253
static drizzle_sys_var* system_variables[]= {
254
DRIZZLE_SYSVAR(enable),
255
DRIZZLE_SYSVAR(hostname),
256
DRIZZLE_SYSVAR(port),
257
DRIZZLE_SYSVAR(username),
258
DRIZZLE_SYSVAR(password),
259
DRIZZLE_SYSVAR(virtualhost),
260
DRIZZLE_SYSVAR(exchange),
261
DRIZZLE_SYSVAR(routingkey),
262
DRIZZLE_SYSVAR(use_replicator),
266
DRIZZLE_PLUGIN(init, system_variables);