← Back to branch summary

~ubuntu-branches/ubuntu/quantal/drizzle/quantal

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

  • Committer: Package Import Robot
  • Author(s): Clint Byrum
  • Date: 2012-06-19 10:46:49 UTC
  • mfrom: (1.2.11) (2.1.16 sid)
  • Revision ID: package-import@ubuntu.com-20120619104649-9ij634mxm4x8pp4l
Tags: 1:7.1.36-stable-1ubuntu1
https://launchpad.net/bugs/987575
* Merge from Debian unstable. (LP: #987575)
  Remaining changes:
  - Added upstart script.
* debian/drizzle.upstart: dropped logger since upstart logs job
  output now.

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/rabbitmq/rabbitmq_log.cc
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)
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(error::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, "%s", 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
  {
120
133
  }
121
134
  try 
122
135
  {
123
 
    rabbitmqLogger= new RabbitMQLog("rabbit_log_applier",
124
 
                                    vm["exchange"].as<string>(),
125
 
                                    vm["routingkey"].as<string>(),
126
 
                                    rabbitmqHandler);
 
136
    rabbitmqLogger= new RabbitMQLog("rabbitmq_applier", rabbitmqHandler);
127
137
  } 
128
138
  catch (exception& e) 
129
139
  {
133
143
  }
134
144
 
135
145
  context.add(rabbitmqLogger);
136
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
137
 
  replication_services.attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
 
146
  ReplicationServices::attachApplier(rabbitmqLogger, vm["use-replicator"].as<string>());
138
147
 
139
148
  context.registerVariable(new sys_var_const_string_val("host", vm["host"].as<string>()));
140
149
  context.registerVariable(new sys_var_constrained_value_readonly<in_port_t>("port", sysvar_rabbitmq_port));
178
187
 
179
188
} /* namespace drizzle_plugin */
180
189
 
181
 
DRIZZLE_PLUGIN(drizzle_plugin::init, NULL, drizzle_plugin::init_options);
182
 
 
 
190
DRIZZLE_DECLARE_PLUGIN
 
191
{
 
192
  DRIZZLE_VERSION_ID,
 
193
  "rabbitmq",
 
194
  "0.1",
 
195
  "Marcus Eriksson",
 
196
  N_("Publishes transactions to RabbitMQ"),
 
197
  PLUGIN_LICENSE_GPL,
 
198
  drizzle_plugin::init,
 
199
  NULL,
 
200
  drizzle_plugin::init_options
 
201
}
 
202
DRIZZLE_DECLARE_PLUGIN_END;