« back to all changes in this revision
Viewing changes to plugin/rabbitmq/rabbitmq_log.cc
- browse files at revision 1283.2.1
- compare with another revision
- compare with revision 2306
- download diff
- download tarball
- view history from revision 1283.2.1
- Committer: Marcus Eriksson
- Date: 2010-02-06 20:18:39 UTC
- mto: This revision was merged to the branch mainline in revision 1438.
- Revision ID: marcuse@localhost.localdomain-20100206201839-7fh52959ox940hud
rabbitmq replication applier
added
removed
plugin/rabbitmq/rabbitmq_log.cc
1
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
2
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
3
*/
4
#include "config.h"
5
#include "rabbitmq_log.h"
6
#include <drizzled/message/transaction.pb.h>
7
#include <google/protobuf/io/coded_stream.h>
8
#include <stdio.h>
9
#include <drizzled/plugin/registry.h>
10
#include <drizzled/plugin.h>
11
#include <stdint.h>
12
#include "rabbitmq_handler.h"
13
14
using namespace std;
15
using namespace drizzled;
16
using namespace google;
17
18
/**
19
* The hostname to connect to
20
*/
21
static char* sysvar_rabbitmq_host= NULL;
22
23
/**
24
* rabbitmq port
25
*/
26
static int sysvar_rabbitmq_port= 0;
27
28
/**
29
* rabbitmq username
30
*/
31
static char* sysvar_rabbitmq_username= NULL;
32
33
/**
34
* rabbitmq password
35
*/
36
static char* sysvar_rabbitmq_password= NULL;
37
38
/**
39
* rabbitmq virtualhost
40
*/
41
static char* sysvar_rabbitmq_virtualhost= NULL;
42
43
/**
44
* rabbitmq exchangename
45
*/
46
static char* sysvar_rabbitmq_exchange= NULL;
47
48
/**
49
* rabbitmq routing key
50
*/
51
static char* sysvar_rabbitmq_routingkey= NULL;
52
53
/**
54
* Is the rabbitmq log enabled?
55
*/
56
static bool sysvar_rabbitmq_log_enabled= false;
57
58
59
RabbitMQLog::RabbitMQLog(const string name_arg,
60
RabbitMQHandler* mqHandler)
61
:plugin::TransactionApplier(name_arg)
62
{
63
rabbitMQHandler= mqHandler;
64
}
65
66
RabbitMQLog::~RabbitMQLog()
67
{
68
}
69
70
void RabbitMQLog::apply(const message::Transaction &to_apply)
71
{
72
size_t message_byte_length= to_apply.ByteSize();
73
uint8_t* buffer= static_cast<uint8_t *>(malloc(message_byte_length));
74
if(buffer == NULL)
75
{
76
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
77
deactivate();
78
return;
79
}
80
81
to_apply.SerializeWithCachedSizesToArray(buffer);
82
try
83
{
84
rabbitMQHandler->publish(buffer,
85
int(message_byte_length),
86
sysvar_rabbitmq_exchange,
87
sysvar_rabbitmq_routingkey);
88
}
89
catch(exception& e)
90
{
91
errmsg_printf(ERRMSG_LVL_ERROR, _(e.what()));
92
deactivate();
93
}
94
free(buffer);
95
}
96
97
RabbitMQLog *rabbitmqLogger; ///< the actual plugin
98
RabbitMQHandler* rabbitmqHandler; ///< the rabbitmq handler
99
100
/**
101
* Initialize the rabbitmq logger - instanciates the dependencies (the handler)
102
* and creates the log handler with the dependency - makes it easier to swap out
103
* handler implementation
104
*/
105
static int init(plugin::Registry ®istry)
106
{
107
if(sysvar_rabbitmq_log_enabled)
108
{
109
try
110
{
111
rabbitmqHandler= new RabbitMQHandler(sysvar_rabbitmq_host,
112
sysvar_rabbitmq_port,
113
sysvar_rabbitmq_username,
114
sysvar_rabbitmq_password,
115
sysvar_rabbitmq_virtualhost);
116
}
117
catch (exception& e)
118
{
119
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler. Got error: %s\n"),
120
e.what());
121
return 1;
122
}
123
try
124
{
125
rabbitmqLogger= new RabbitMQLog("rabbit-log", rabbitmqHandler);
126
}
127
catch (exception& e)
128
{
129
errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance. Got error: %s\n"),
130
e.what());
131
return 1;
132
}
133
134
registry.add(rabbitmqLogger);
135
return 0;
136
}
137
return 0;
138
}
139
140
static int deinit(plugin::Registry ®istry)
141
{
142
/* Cleanup the logger itself - delete the logger first, then handler, to avoid NPEs */
143
if (rabbitmqLogger)
144
{
145
registry.remove(rabbitmqLogger);
146
delete rabbitmqLogger;
147
delete rabbitmqHandler;
148
}
149
150
return 0;
151
}
152
153
154
155
static DRIZZLE_SYSVAR_BOOL(enable,
156
sysvar_rabbitmq_log_enabled,
157
PLUGIN_VAR_NOCMDARG,
158
N_("Enable rabbitmq log"),
159
NULL, /* check func */
160
NULL, /* update func */
161
false /* default */);
162
163
164
static DRIZZLE_SYSVAR_STR(hostname,
165
sysvar_rabbitmq_host,
166
PLUGIN_VAR_READONLY,
167
N_("Host name to connect to"),
168
NULL, /* check func */
169
NULL, /* update func*/
170
"localhost" /* default */);
171
172
173
static DRIZZLE_SYSVAR_INT(port,
174
sysvar_rabbitmq_port,
175
PLUGIN_VAR_READONLY,
176
N_("RabbitMQ Port"),
177
NULL, /* check func */
178
NULL, /* update func */
179
5672, /* default */
180
0,
181
65535,
182
0);
183
184
static DRIZZLE_SYSVAR_STR(username,
185
sysvar_rabbitmq_username,
186
PLUGIN_VAR_READONLY,
187
N_("RabbitMQ username"),
188
NULL, /* check func */
189
NULL, /* update func*/
190
"guest" /* default */);
191
192
static DRIZZLE_SYSVAR_STR(password,
193
sysvar_rabbitmq_password,
194
PLUGIN_VAR_READONLY,
195
N_("RabbitMQ password"),
196
NULL, /* check func */
197
NULL, /* update func*/
198
"guest" /* default */);
199
200
static DRIZZLE_SYSVAR_STR(virtualhost,
201
sysvar_rabbitmq_virtualhost,
202
PLUGIN_VAR_READONLY,
203
N_("RabbitMQ virtualhost"),
204
NULL, /* check func */
205
NULL, /* update func*/
206
"/" /* default */);
207
208
static DRIZZLE_SYSVAR_STR(exchange,
209
sysvar_rabbitmq_exchange,
210
PLUGIN_VAR_READONLY,
211
N_("Name of RabbitMQ exchange to publish to"),
212
NULL, /* check func */
213
NULL, /* update func*/
214
"ReplicationExchange" /* default */);
215
216
static DRIZZLE_SYSVAR_STR(routingkey,
217
sysvar_rabbitmq_routingkey,
218
PLUGIN_VAR_READONLY,
219
N_("Name of RabbitMQ routing key to use"),
220
NULL, /* check func */
221
NULL, /* update func*/
222
"ReplicationRoutingKey" /* default */);
223
224
225
static drizzle_sys_var* system_variables[]= {
226
DRIZZLE_SYSVAR(enable),
227
DRIZZLE_SYSVAR(hostname),
228
DRIZZLE_SYSVAR(port),
229
DRIZZLE_SYSVAR(username),
230
DRIZZLE_SYSVAR(password),
231
DRIZZLE_SYSVAR(virtualhost),
232
DRIZZLE_SYSVAR(exchange),
233
DRIZZLE_SYSVAR(routingkey),
234
NULL
235
};
236
237
DRIZZLE_PLUGIN(init, deinit, NULL, system_variables);
238
Loggerhead is a web-based interface for Breezy
Version: 2.0.1