← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/rabbitmq/rabbitmq_log.cc

  • Committer: pcrews
  • Date: 2011-05-24 17:36:24 UTC
  • mfrom: (1099.4.232 drizzle)
  • Revision ID: pcrews@lucid32-20110524173624-mwr1bvq6fa1r01ao
Updated translations + 2011.05.18 tarball tag

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)
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, _(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
  {
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)