← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

  • Committer: Mark Atwood
  • Date: 2011-10-27 05:08:12 UTC
  • mfrom: (2445.1.11 rf)
  • Revision ID: me@mark.atwood.name-20111027050812-1icvs72lb0u4xdc4
merge lp:~olafvdspek/drizzle/refactor8

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/rabbitmq/rabbitmq_log.cc
22
22
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
23
23
 */
24
24
 
25
 
#include "config.h"
 
25
#include <config.h>
26
26
#include "rabbitmq_log.h"
27
27
#include <drizzled/message/transaction.pb.h>
28
28
#include <google/protobuf/io/coded_stream.h>
50
50
 
51
51
 
52
52
RabbitMQLog::RabbitMQLog(const string &name, 
53
 
                         const std::string &exchange,
54
 
                         const std::string &routingkey,
55
53
                         RabbitMQHandler* mqHandler) :
56
54
  plugin::TransactionApplier(name),
57
 
  _rabbitMQHandler(mqHandler),
58
 
  _exchange(exchange),
59
 
  _routingkey(routingkey)
 
55
  _rabbitMQHandler(mqHandler)
60
56
{ }
61
57
 
62
58
RabbitMQLog::~RabbitMQLog() 
63
 
{ }
 
59
{ 
 
60
  _rabbitMQHandler->disconnect();
 
61
  delete _rabbitMQHandler;
 
62
}
64
63
 
65
64
plugin::ReplicationReturnCode
66
65
RabbitMQLog::apply(Session &, const message::Transaction &to_apply)
69
68
  uint8_t* buffer= new uint8_t[message_byte_length];
70
69
  if(buffer == NULL)
71
70
  {
72
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate enough memory to transaction message\n"));
 
71
    errmsg_printf(error::ERROR, _("Failed to allocate enough memory to transaction message\n"));
73
72
    deactivate();
74
73
    return plugin::UNKNOWN_ERROR;
75
74
  }
76
75
 
77
76
  to_apply.SerializeWithCachedSizesToArray(buffer);
78
 
  try
79
 
  {
80
 
    _rabbitMQHandler->publish(buffer, 
81
 
                             int(message_byte_length), 
82
 
                             _exchange,
83
 
                             _routingkey);
84
 
  }
85
 
  catch(exception& e)
86
 
  {
87
 
    errmsg_printf(ERRMSG_LVL_ERROR, _(e.what()));
88
 
    deactivate();
89
 
    return plugin::UNKNOWN_ERROR;
90
 
  }
 
77
  short tries = 3;
 
78
  bool sent = false;
 
79
  while (!sent && tries > 0) {
 
80
    tries--;
 
81
    try
 
82
    {
 
83
      _rabbitMQHandler->publish(buffer, int(message_byte_length));
 
84
      sent = true;
 
85
    } 
 
86
    catch(exception& e)
 
87
    {
 
88
      errmsg_printf(error::ERROR, _(e.what()));
 
89
      try {
 
90
        _rabbitMQHandler->reconnect();
 
91
      } catch(exception &e) {
 
92
        errmsg_printf(error::ERROR, _("Could not reconnect, trying again.. - waiting 10 seconds for server to come back"));
 
93
        sleep(10);
 
94
      } // 
 
95
    }
 
96
  }
 
97
 
91
98
  delete[] buffer;
92
 
  return plugin::SUCCESS;
 
99
  if(sent) return plugin::SUCCESS;
 
100
  errmsg_printf(error::ERROR, _("RabbitMQ server has disappeared, failing transaction."));
 
101
  deactivate();
 
102
  return plugin::UNKNOWN_ERROR;
93
103
}
94
104
 
95
105
static RabbitMQLog *rabbitmqLogger; ///< the actual plugin
96
106
static RabbitMQHandler* rabbitmqHandler; ///< the rabbitmq handler
97
107
 
 
108
 
98
109
/**
99
110
 * Initialize the rabbitmq logger - instanciates the dependencies (the handler)
100
111
 * and creates the log handler with the dependency - makes it easier to swap out
110
121
                                         sysvar_rabbitmq_port, 
111
122
                                         vm["username"].as<string>(), 
112
123
                                         vm["password"].as<string>(), 
113
 
                                         vm["virtualhost"].as<string>());
 
124
                                         vm["virtualhost"].as<string>(),
 
125
                                         vm["exchange"].as<string>(),
 
126
                                         vm["routingkey"].as<string>());
114
127
  } 
115
128
  catch (exception& e) 
116
129
  {
117
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
 
130
    errmsg_printf(error::ERROR, _("Failed to allocate the RabbitMQHandler.  Got error: %s\n"),
118
131
                  e.what());
119
132
    return 1;
120
133
  }
121
134
  try 
122
135
  {
123
136
    rabbitmqLogger= new RabbitMQLog("rabbit_log_applier",
124
 
                                    vm["exchange"].as<string>(),
125
 
                                    vm["routingkey"].as<string>(),
126
137
                                    rabbitmqHandler);
127
138
  } 
128
139
  catch (exception& e) 
129
140
  {
130
 
    errmsg_printf(ERRMSG_LVL_ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
 
141
    errmsg_printf(error::ERROR, _("Failed to allocate the RabbitMQLog instance.  Got error: %s\n"), 
131
142
                  e.what());
132
143
    return 1;
133
144
  }
134
145
 
135
146
  context.add(rabbitmqLogger);
136
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
137
 
  replication_services.attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
 
147
  ReplicationServices::attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
138
148
 
139
149
  context.registerVariable(new sys_var_const_string_val("host", vm["host"].as<string>()));
140
150
  context.registerVariable(new sys_var_constrained_value_readonly<in_port_t>("port", sysvar_rabbitmq_port));
152
162
{
153
163
  context("host", 
154
164
          po::value<string>()->default_value("localhost"),
155
 
          N_("Host name to connect to"));
 
165
          _("Host name to connect to"));
156
166
  context("port",
157
167
          po::value<port_constraint>(&sysvar_rabbitmq_port)->default_value(5672),
158
 
          N_("Port to connect to"));
 
168
          _("Port to connect to"));
159
169
  context("virtualhost",
160
170
          po::value<string>()->default_value("/"),
161
 
          N_("RabbitMQ virtualhost"));
 
171
          _("RabbitMQ virtualhost"));
162
172
  context("username",
163
173
          po::value<string>()->default_value("guest"),
164
 
          N_("RabbitMQ username"));
 
174
          _("RabbitMQ username"));
165
175
  context("password",
166
176
          po::value<string>()->default_value("guest"),
167
 
          N_("RabbitMQ password"));
 
177
          _("RabbitMQ password"));
168
178
  context("use-replicator",
169
179
          po::value<string>()->default_value("default_replicator"),
170
 
          N_("Name of the replicator plugin to use (default='default_replicator')"));
 
180
          _("Name of the replicator plugin to use (default='default_replicator')"));
171
181
  context("exchange",
172
182
          po::value<string>()->default_value("ReplicationExchange"),
173
 
          N_("Name of RabbitMQ exchange to publish to"));
 
183
          _("Name of RabbitMQ exchange to publish to"));
174
184
  context("routingkey",
175
185
          po::value<string>()->default_value("ReplicationRoutingKey"),
176
 
          N_("Name of RabbitMQ routing key to use"));
 
186
          _("Name of RabbitMQ routing key to use"));
177
187
}
178
188
 
179
189
} /* namespace drizzle_plugin */