← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/slave/queue_consumer.h

  • Committer: Mark Atwood
  • Date: 2011-08-11 03:05:03 UTC
  • mfrom: (2385.1.12 refactor4)
  • Revision ID: me@mark.atwood.name-20110811030503-rp9xjihc5x3y0x4q
merge lp:~olafvdspek/drizzle/refactor4

Show diffs side-by-side

added added

removed removed

Lines of Context:
plugin/slave/queue_consumer.h
 
1
/* - mode: c; c-basic-offset: 2; indent-tabs-mode: nil; -*-
 
2
 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
 
3
 *
 
4
 *  Copyright (C) 2011 David Shrewsbury
 
5
 *
 
6
 *  This program is free software; you can redistribute it and/or modify
 
7
 *  it under the terms of the GNU General Public License as published by
 
8
 *  the Free Software Foundation; either version 2 of the License, or
 
9
 *  (at your option) any later version.
 
10
 *
 
11
 *  This program is distributed in the hope that it will be useful,
 
12
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
 *  GNU General Public License for more details.
 
15
 *
 
16
 *  You should have received a copy of the GNU General Public License
 
17
 *  along with this program; if not, write to the Free Software
 
18
 *  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 
19
 */
 
20
 
 
21
#pragma once
 
22
 
 
23
#include <plugin/slave/queue_thread.h>
 
24
#include <plugin/slave/sql_executor.h>
 
25
#include <drizzled/session.h>
 
26
#include <string>
 
27
 
 
28
namespace drizzled
 
29
{
 
30
  namespace message
 
31
  {
 
32
    class Transaction;
 
33
  }
 
34
}
 
35
 
 
36
namespace slave
 
37
{
 
38
 
 
39
class QueueConsumer : public QueueThread, public SQLExecutor
 
40
{
 
41
public:
 
42
  QueueConsumer() :
 
43
    QueueThread(),
 
44
    SQLExecutor("slave", "replication"),
 
45
    _check_interval(5),
 
46
    _ignore_errors(false)
 
47
  { }
 
48
 
 
49
  bool init();
 
50
  bool process();
 
51
  void shutdown();
 
52
 
 
53
  void setSleepInterval(uint32_t seconds)
 
54
  {
 
55
    _check_interval= seconds;
 
56
  }
 
57
 
 
58
  uint32_t getSleepInterval()
 
59
  {
 
60
    return _check_interval;
 
61
  }
 
62
  
 
63
  /**
 
64
   * Determines if we should ignore errors from statements pulled from masters.
 
65
   */
 
66
  void setIgnoreErrors(bool value)
 
67
  {
 
68
    _ignore_errors= value;
 
69
  }
 
70
 
 
71
  /**
 
72
   * Update applier status in state table.
 
73
   *
 
74
   * @param err_msg Error message string
 
75
   * @param status false = STOPPED, true = RUNNING
 
76
   */
 
77
  void setApplierState(const std::string &err_msg, bool status);
 
78
 
 
79
  void addMasterId(uint32_t id)
 
80
  {
 
81
    _master_ids.push_back(id);
 
82
  }
 
83
 
 
84
  bool processSingleMaster(const std::string &master_id);
 
85
 
 
86
private:
 
87
  typedef std::vector<uint64_t> TrxIdList;
 
88
 
 
89
  /** Number of seconds to sleep between checking queue for messages */
 
90
  uint32_t _check_interval;
 
91
 
 
92
  std::vector<uint32_t> _master_ids;
 
93
 
 
94
  bool _ignore_errors;
 
95
 
 
96
  /**
 
97
   * Get a list of transaction IDs from the queue that are complete.
 
98
   *
 
99
   * A "complete" transaction is one in which we have received the end
 
100
   * segment of the transaction.
 
101
   *
 
102
   * @param[in] master_id Identifier of the master we are interested in.
 
103
   * @param[out] list The list to populate with transaction IDs.
 
104
   *
 
105
   * @retval true Success
 
106
   * @retval false Error
 
107
   */
 
108
  bool getListOfCompletedTransactions(const std::string &master_id,
 
109
                                      TrxIdList &list);
 
110
 
 
111
  bool getMessage(drizzled::message::Transaction &transaction,
 
112
                  std::string &commit_id,
 
113
                  const std::string &master_id,
 
114
                  uint64_t trx_id,
 
115
                  std::string &originating_server_uuid,
 
116
                  uint64_t &originating_commit_id,
 
117
                  uint32_t segment_id);
 
118
 
 
119
  /**
 
120
   * Convert the given Transaction message into equivalent SQL.
 
121
   *
 
122
   * @param[in] transaction Transaction protobuf message to convert.
 
123
   * @param[in,out] aggregate_sql Buffer for total SQL for this transaction.
 
124
   * @param[in,out] segmented_sql Buffer for carried over segmented statements.
 
125
   *
 
126
   * @retval true Success
 
127
   * @retval false Failure
 
128
   */
 
129
  bool convertToSQL(const drizzled::message::Transaction &transaction,
 
130
                    std::vector<std::string> &aggregate_sql,
 
131
                    std::vector<std::string> &segmented_sql);
 
132
 
 
133
  /**
 
134
   * Execute a batch of SQL statements.
 
135
   *
 
136
   * @param sql Batch of SQL statements to execute.
 
137
   * @param commit_id Commit ID value to store in state table.
 
138
   * @param originating_server_uuid Server ID of the master where
 
139
   *   this SQL was originally applied.
 
140
   * @param originating_commit_id Commit ID of the master where
 
141
   *   this SQL was originally applied.
 
142
   *
 
143
   * @retval true Success
 
144
   * @retval false Failure
 
145
   */
 
146
  bool executeSQLWithCommitId(std::vector<std::string> &sql,
 
147
                              const std::string &commit_id,
 
148
                              const std::string &originating_server_uuid,
 
149
                              uint64_t originating_commit_id,
 
150
                              const std::string &master_id);
 
151
  
 
152
  /**
 
153
   * Remove messages for a given transaction from the queue.
 
154
   *
 
155
   * @param trx_id Transaction ID for the messages to remove.
 
156
   *
 
157
   * @retval true Success
 
158
   * @retval false Failure
 
159
   */
 
160
  bool deleteFromQueue(const std::string &master_id, uint64_t trx_id);
 
161
 
 
162
  /**
 
163
   * Determine if a Statement message is an end message.
 
164
   *
 
165
   * @retval true Is an end Statement message
 
166
   * @retval false Is NOT an end Statement message
 
167
   */
 
168
  bool isEndStatement(const drizzled::message::Statement &statement);
 
169
};
 
170
 
 
171
} /* namespace slave */
 
172