← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

  • Committer: Brian Aker
  • Date: 2010-06-28 16:17:36 UTC
  • mfrom: (1637.4.1 drizzle)
  • Revision ID: brian@gaz-20100628161736-eormhb2mnd551i2h
Merge unused

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/rabbitmq/rabbitmq_log.cc
31
31
#include <drizzled/plugin.h>
32
32
#include <stdint.h>
33
33
#include "rabbitmq_handler.h"
34
 
#include <boost/program_options.hpp>
35
 
#include <drizzled/module/option_map.h>
36
 
 
37
 
namespace po= boost::program_options;
38
34
 
39
35
using namespace std;
40
36
using namespace drizzled;
41
37
using namespace google;
42
38
 
43
 
namespace drizzle_plugin
44
 
{
 
39
/**
 
40
 * The hostname to connect to
 
41
 */
 
42
static char* sysvar_rabbitmq_host= NULL;
45
43
 
46
44
/**
47
45
 * rabbitmq port
48
46
 */
49
 
static port_constraint sysvar_rabbitmq_port;
50
 
 
51
 
 
52
 
RabbitMQLog::RabbitMQLog(const string &name, 
53
 
                         const std::string &exchange,
54
 
                         const std::string &routingkey,
55
 
                         RabbitMQHandler* mqHandler) :
56
 
  plugin::TransactionApplier(name),
57
 
  _rabbitMQHandler(mqHandler),
58
 
  _exchange(exchange),
59
 
  _routingkey(routingkey)
60
 
{ }
 
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
 * The name of the replicator plugin
 
81
 * to pair the rabbitmq log's applier with.
 
82
 * Defaults to "default"
 
83
 */
 
84
static char *sysvar_rabbitmq_use_replicator= NULL;
 
85
static const char DEFAULT_USE_REPLICATOR[]= "default";
 
86
 
 
87
 
 
88
RabbitMQLog::RabbitMQLog(const string name_arg, 
 
89
                         RabbitMQHandler* mqHandler)
 
90
  :plugin::TransactionApplier(name_arg)
 
91
{
 
92
  rabbitMQHandler= mqHandler;
 
93
}
61
94
 
62
95
RabbitMQLog::~RabbitMQLog() 
63
 
{ }
 
96
{
 
97
}
64
98
 
65
99
plugin::ReplicationReturnCode
66
100
RabbitMQLog::apply(Session &, const message::Transaction &to_apply)
67
101
{
68
102
  size_t message_byte_length= to_apply.ByteSize();
69
 
  uint8_t* buffer= new uint8_t[message_byte_length];
 
103
  uint8_t* buffer= static_cast<uint8_t *>(malloc(message_byte_length));
70
104
  if(buffer == NULL)
71
105
  {
72
 
    errmsg_printf(error::ERROR, _("Failed to allocate enough memory to transaction message\n"));
 
106
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
73
107
    deactivate();
74
108
    return plugin::UNKNOWN_ERROR;
75
109
  }
77
111
  to_apply.SerializeWithCachedSizesToArray(buffer);
78
112
  try
79
113
  {
80
 
    _rabbitMQHandler->publish(buffer, 
81
 
                             int(message_byte_length), 
82
 
                             _exchange,
83
 
                             _routingkey);
 
114
    rabbitMQHandler->publish(buffer, 
 
115
                             int(message_byte_length), 
 
116
                             sysvar_rabbitmq_exchange, 
 
117
                             sysvar_rabbitmq_routingkey);
84
118
  }
85
119
  catch(exception& e)
86
120
  {
87
 
    errmsg_printf(error::ERROR, _(e.what()));
 
121
    errmsg_printf(ERRMSG_LVL_ERROR, _(e.what()));
88
122
    deactivate();
89
123
    return plugin::UNKNOWN_ERROR;
90
124
  }
91
 
  delete[] buffer;
 
125
  free(buffer);
92
126
  return plugin::SUCCESS;
93
127
}
94
128
 
102
136
 */
103
137
static int init(drizzled::module::Context &context)
104
138
{
105
 
  const module::option_map &vm= context.getOptions();
106
 
  
107
 
  try 
108
 
  {
109
 
    rabbitmqHandler= new RabbitMQHandler(vm["host"].as<string>(),
110
 
                                         sysvar_rabbitmq_port, 
111
 
                                         vm["username"].as<string>(), 
112
 
                                         vm["password"].as<string>(), 
113
 
                                         vm["virtualhost"].as<string>());
114
 
  } 
115
 
  catch (exception& e) 
116
 
  {
117
 
    errmsg_printf(error::ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
118
 
                  e.what());
119
 
    return 1;
120
 
  }
121
 
  try 
122
 
  {
123
 
    rabbitmqLogger= new RabbitMQLog("rabbit_log_applier",
124
 
                                    vm["exchange"].as<string>(),
125
 
                                    vm["routingkey"].as<string>(),
126
 
                                    rabbitmqHandler);
127
 
  } 
128
 
  catch (exception& e) 
129
 
  {
130
 
    errmsg_printf(error::ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
131
 
                  e.what());
132
 
    return 1;
133
 
  }
134
 
 
135
 
  context.add(rabbitmqLogger);
136
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
137
 
  replication_services.attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
138
 
 
139
 
  context.registerVariable(new sys_var_const_string_val("host", vm["host"].as<string>()));
140
 
  context.registerVariable(new sys_var_constrained_value_readonly<in_port_t>("port", sysvar_rabbitmq_port));
141
 
  context.registerVariable(new sys_var_const_string_val("username", vm["username"].as<string>()));
142
 
  context.registerVariable(new sys_var_const_string_val("password", vm["password"].as<string>()));
143
 
  context.registerVariable(new sys_var_const_string_val("virtualhost", vm["virtualhost"].as<string>()));
144
 
  context.registerVariable(new sys_var_const_string_val("exchange", vm["exchange"].as<string>()));
145
 
  context.registerVariable(new sys_var_const_string_val("routingkey", vm["routingkey"].as<string>()));
146
 
 
 
139
  if(sysvar_rabbitmq_log_enabled)
 
140
  {
 
141
    try 
 
142
    {
 
143
      rabbitmqHandler= new RabbitMQHandler(sysvar_rabbitmq_host, 
 
144
                                           sysvar_rabbitmq_port, 
 
145
                                           sysvar_rabbitmq_username, 
 
146
                                           sysvar_rabbitmq_password, 
 
147
                                           sysvar_rabbitmq_virtualhost);
 
148
    } 
 
149
    catch (exception& e) 
 
150
    {
 
151
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
 
152
                    e.what());
 
153
      return 1;
 
154
    }
 
155
    try 
 
156
    {
 
157
      rabbitmqLogger= new RabbitMQLog("rabbit_log_applier", rabbitmqHandler);
 
158
    } 
 
159
    catch (exception& e) 
 
160
    {
 
161
      errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
 
162
                    e.what());
 
163
      return 1;
 
164
    }
 
165
 
 
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);
 
170
    return 0;
 
171
  }
147
172
  return 0;
148
173
}
149
174
 
150
175
 
151
 
static void init_options(drizzled::module::option_context &context)
152
 
{
153
 
  context("host", 
154
 
          po::value<string>()->default_value("localhost"),
155
 
          _("Host name to connect to"));
156
 
  context("port",
157
 
          po::value<port_constraint>(&sysvar_rabbitmq_port)->default_value(5672),
158
 
          _("Port to connect to"));
159
 
  context("virtualhost",
160
 
          po::value<string>()->default_value("/"),
161
 
          _("RabbitMQ virtualhost"));
162
 
  context("username",
163
 
          po::value<string>()->default_value("guest"),
164
 
          _("RabbitMQ username"));
165
 
  context("password",
166
 
          po::value<string>()->default_value("guest"),
167
 
          _("RabbitMQ password"));
168
 
  context("use-replicator",
169
 
          po::value<string>()->default_value("default_replicator"),
170
 
          _("Name of the replicator plugin to use (default='default_replicator')"));
171
 
  context("exchange",
172
 
          po::value<string>()->default_value("ReplicationExchange"),
173
 
          _("Name of RabbitMQ exchange to publish to"));
174
 
  context("routingkey",
175
 
          po::value<string>()->default_value("ReplicationRoutingKey"),
176
 
          _("Name of RabbitMQ routing key to use"));
177
 
}
178
 
 
179
 
} /* namespace drizzle_plugin */
180
 
 
181
 
DRIZZLE_PLUGIN(drizzle_plugin::init, NULL, drizzle_plugin::init_options);
 
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
static DRIZZLE_SYSVAR_STR(use_replicator,
 
246
                          sysvar_rabbitmq_use_replicator,
 
247
                          PLUGIN_VAR_READONLY,
 
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 */);
 
252
 
 
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),
 
263
  NULL
 
264
};
 
265
 
 
266
DRIZZLE_PLUGIN(init, system_variables);
182
267