← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

  • Committer: Stewart Smith
  • Date: 2010-04-06 07:38:20 UTC
  • mfrom: (1283.28.40)
  • mto: (1283.38.1)
  • mto: This revision was merged to the branch mainline in revision 1536.
  • Revision ID: stewart@flamingspork.com-20100406073820-uf1mqht2jhn5zyl4
Merged embedded-innodb-position-rnd-pos into embedded-innodb-autoincrement.

Show diffs side-by-side

added added

removed removed

Lines of Context:
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
 *  Copyright (C) 2010 Marcus Eriksson
 
5
 *
 
6
 *  Authors:
 
7
 *
 
8
 *  Marcus Eriksson <krummas@gmail.com>
 
9
 *
 
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.
 
14
 *
 
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.
 
19
 *
 
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
 
23
 */
 
24
 
 
25
#include "config.h"
 
26
#include "rabbitmq_log.h"
 
27
#include <drizzled/message/transaction.pb.h>
 
28
#include <google/protobuf/io/coded_stream.h>
 
29
#include <stdio.h>
 
30
#include <drizzled/plugin/registry.h>
 
31
#include <drizzled/plugin.h>
 
32
#include <stdint.h>
 
33
#include "rabbitmq_handler.h"
 
34
 
 
35
using namespace std;
 
36
using namespace drizzled;
 
37
using namespace google;
 
38
 
 
39
/**
 
40
 * The hostname to connect to
 
41
 */
 
42
static char* sysvar_rabbitmq_host= NULL;
 
43
 
 
44
/**
 
45
 * rabbitmq port
 
46
 */
 
47
static int sysvar_rabbitmq_port= 0;
 
48
 
 
49
/**
 
50
 * rabbitmq username
 
51
 */
 
52
static char* sysvar_rabbitmq_username= NULL;
 
53
 
 
54
/**
 
55
 * rabbitmq password
 
56
 */
 
57
static char* sysvar_rabbitmq_password= NULL;
 
58
 
 
59
/**
 
60
 * rabbitmq virtualhost
 
61
 */
 
62
static char* sysvar_rabbitmq_virtualhost= NULL;
 
63
 
 
64
/**
 
65
 * rabbitmq exchangename
 
66
 */
 
67
static char* sysvar_rabbitmq_exchange= NULL;
 
68
 
 
69
/**
 
70
 * rabbitmq routing key
 
71
 */
 
72
static char* sysvar_rabbitmq_routingkey= NULL;
 
73
 
 
74
/**
 
75
 * Is the rabbitmq log enabled?
 
76
 */
 
77
static bool sysvar_rabbitmq_log_enabled= false;
 
78
 
 
79
 
 
80
RabbitMQLog::RabbitMQLog(const string name_arg, 
 
81
                         RabbitMQHandler* mqHandler)
 
82
  :plugin::TransactionApplier(name_arg)
 
83
{
 
84
  rabbitMQHandler= mqHandler;
 
85
}
 
86
 
 
87
RabbitMQLog::~RabbitMQLog() 
 
88
{
 
89
}
 
90
 
 
91
void RabbitMQLog::apply(const message::Transaction &to_apply)
 
92
{
 
93
  size_t message_byte_length= to_apply.ByteSize();
 
94
  uint8_t* buffer= static_cast<uint8_t *>(malloc(message_byte_length));
 
95
  if(buffer == NULL)
 
96
  {
 
97
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
 
98
    deactivate();
 
99
    return;
 
100
  }
 
101
 
 
102
  to_apply.SerializeWithCachedSizesToArray(buffer);
 
103
  try
 
104
  {
 
105
    rabbitMQHandler->publish(buffer, 
 
106
                             int(message_byte_length), 
 
107
                             sysvar_rabbitmq_exchange, 
 
108
                             sysvar_rabbitmq_routingkey);
 
109
  }
 
110
  catch(exception& e)
 
111
  {
 
112
    errmsg_printf(ERRMSG_LVL_ERROR, _(e.what()));
 
113
    deactivate();
 
114
  }
 
115
  free(buffer);
 
116
}
 
117
 
 
118
RabbitMQLog *rabbitmqLogger; ///< the actual plugin
 
119
RabbitMQHandler* rabbitmqHandler; ///< the rabbitmq handler
 
120
 
 
121
/**
 
122
 * Initialize the rabbitmq logger - instanciates the dependencies (the handler)
 
123
 * and creates the log handler with the dependency - makes it easier to swap out
 
124
 * handler implementation
 
125
 */
 
126
static int init(plugin::Registry &registry)
 
127
{
 
128
  if(sysvar_rabbitmq_log_enabled)
 
129
  {
 
130
    try 
 
131
    {
 
132
      rabbitmqHandler= new RabbitMQHandler(sysvar_rabbitmq_host, 
 
133
                                           sysvar_rabbitmq_port, 
 
134
                                           sysvar_rabbitmq_username, 
 
135
                                           sysvar_rabbitmq_password, 
 
136
                                           sysvar_rabbitmq_virtualhost);
 
137
    } 
 
138
    catch (exception& e) 
 
139
    {
 
140
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
 
141
                    e.what());
 
142
      return 1;
 
143
    }
 
144
    try 
 
145
    {
 
146
      rabbitmqLogger= new RabbitMQLog("rabbit-log", rabbitmqHandler);
 
147
    } 
 
148
    catch (exception& e) 
 
149
    {
 
150
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
 
151
                    e.what());
 
152
      return 1;
 
153
    }
 
154
 
 
155
    registry.add(rabbitmqLogger);
 
156
    return 0;
 
157
  }
 
158
  return 0;
 
159
}
 
160
 
 
161
static int deinit(plugin::Registry &registry)
 
162
{
 
163
  /* Cleanup the logger itself - delete the logger first, then handler, to avoid NPEs */
 
164
  if (rabbitmqLogger)
 
165
  {
 
166
    registry.remove(rabbitmqLogger);
 
167
    delete rabbitmqLogger;
 
168
    delete rabbitmqHandler;
 
169
  }
 
170
 
 
171
  return 0;
 
172
}
 
173
 
 
174
 
 
175
 
 
176
static DRIZZLE_SYSVAR_BOOL(enable,
 
177
                           sysvar_rabbitmq_log_enabled,
 
178
                           PLUGIN_VAR_NOCMDARG,
 
179
                           N_("Enable rabbitmq log"),
 
180
                           NULL, /* check func */
 
181
                           NULL, /* update func */
 
182
                           false /* default */);
 
183
 
 
184
 
 
185
static DRIZZLE_SYSVAR_STR(hostname,
 
186
                          sysvar_rabbitmq_host,
 
187
                          PLUGIN_VAR_READONLY,
 
188
                          N_("Host name to connect to"),
 
189
                          NULL, /* check func */
 
190
                          NULL, /* update func*/
 
191
                          "localhost" /* default */);
 
192
 
 
193
 
 
194
static DRIZZLE_SYSVAR_INT(port,
 
195
                          sysvar_rabbitmq_port,
 
196
                          PLUGIN_VAR_READONLY,
 
197
                          N_("RabbitMQ Port"),
 
198
                          NULL, /* check func */
 
199
                          NULL, /* update func */
 
200
                          5672, /* default */
 
201
                          0,
 
202
                          65535,
 
203
                          0);
 
204
 
 
205
static DRIZZLE_SYSVAR_STR(username,
 
206
                          sysvar_rabbitmq_username,
 
207
                          PLUGIN_VAR_READONLY,
 
208
                          N_("RabbitMQ username"),
 
209
                          NULL, /* check func */
 
210
                          NULL, /* update func*/
 
211
                          "guest" /* default */);
 
212
 
 
213
static DRIZZLE_SYSVAR_STR(password,
 
214
                          sysvar_rabbitmq_password,
 
215
                          PLUGIN_VAR_READONLY,
 
216
                          N_("RabbitMQ password"),
 
217
                          NULL, /* check func */
 
218
                          NULL, /* update func*/
 
219
                          "guest" /* default */);
 
220
 
 
221
static DRIZZLE_SYSVAR_STR(virtualhost,
 
222
                          sysvar_rabbitmq_virtualhost,
 
223
                          PLUGIN_VAR_READONLY,
 
224
                          N_("RabbitMQ virtualhost"),
 
225
                          NULL, /* check func */
 
226
                          NULL, /* update func*/
 
227
                          "/" /* default */);
 
228
 
 
229
static DRIZZLE_SYSVAR_STR(exchange,
 
230
                          sysvar_rabbitmq_exchange,
 
231
                          PLUGIN_VAR_READONLY,
 
232
                          N_("Name of RabbitMQ exchange to publish to"),
 
233
                          NULL, /* check func */
 
234
                          NULL, /* update func*/
 
235
                          "ReplicationExchange" /* default */);
 
236
 
 
237
static DRIZZLE_SYSVAR_STR(routingkey,
 
238
                          sysvar_rabbitmq_routingkey,
 
239
                          PLUGIN_VAR_READONLY,
 
240
                          N_("Name of RabbitMQ routing key to use"),
 
241
                          NULL, /* check func */
 
242
                          NULL, /* update func*/
 
243
                          "ReplicationRoutingKey" /* default */);
 
244
 
 
245
 
 
246
static drizzle_sys_var* system_variables[]= {
 
247
  DRIZZLE_SYSVAR(enable),
 
248
  DRIZZLE_SYSVAR(hostname),
 
249
  DRIZZLE_SYSVAR(port),
 
250
  DRIZZLE_SYSVAR(username),
 
251
  DRIZZLE_SYSVAR(password),
 
252
  DRIZZLE_SYSVAR(virtualhost),
 
253
  DRIZZLE_SYSVAR(exchange),
 
254
  DRIZZLE_SYSVAR(routingkey),
 
255
  NULL
 
256
};
 
257
 
 
258
DRIZZLE_PLUGIN(init, deinit, NULL, system_variables);
 
259