← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to plugin/slave/queue_consumer.h

  • Committer: Brian Aker
  • Date: 2011-02-25 17:02:30 UTC
  • mfrom: (2116.1.55 slave)
  • Revision ID: brian@tangent.org-20110225170230-zj0h32xlmr42ly2z
Merge in David's slave work

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
#ifndef PLUGIN_SLAVE_QUEUE_CONSUMER_H
 
22
#define PLUGIN_SLAVE_QUEUE_CONSUMER_H
 
23
 
 
24
#include <plugin/slave/queue_thread.h>
 
25
#include <plugin/slave/sql_executor.h>
 
26
#include <drizzled/session.h>
 
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
  { }
 
47
 
 
48
  bool init();
 
49
  bool process();
 
50
  void shutdown();
 
51
 
 
52
  void setSleepInterval(uint32_t seconds)
 
53
  {
 
54
    _check_interval= seconds;
 
55
  }
 
56
 
 
57
  uint32_t getSleepInterval()
 
58
  {
 
59
    return _check_interval;
 
60
  }
 
61
  
 
62
  /**
 
63
   * Update applier status in state table.
 
64
   *
 
65
   * @param err_msg Error message string
 
66
   * @param status false = STOPPED, true = RUNNING
 
67
   */
 
68
  void setApplierState(const std::string &err_msg, bool status);
 
69
 
 
70
private:
 
71
  typedef std::vector<uint64_t> TrxIdList;
 
72
 
 
73
  /** Number of seconds to sleep between checking queue for messages */
 
74
  uint32_t _check_interval;
 
75
 
 
76
  bool getListOfCompletedTransactions(TrxIdList &list);
 
77
 
 
78
  bool getMessage(drizzled::message::Transaction &transaction,
 
79
                  std::string &commit_id,
 
80
                  uint64_t trx_id,
 
81
                  uint32_t segment_id);
 
82
 
 
83
  /**
 
84
   * Convert the given Transaction message into equivalent SQL.
 
85
   *
 
86
   * @param[in] transaction Transaction protobuf message to convert.
 
87
   * @param[in,out] aggregate_sql Buffer for total SQL for this transaction.
 
88
   * @param[in,out] segmented_sql Buffer for carried over segmented statements.
 
89
   *
 
90
   * @retval true Success
 
91
   * @retval false Failure
 
92
   */
 
93
  bool convertToSQL(const drizzled::message::Transaction &transaction,
 
94
                    std::vector<std::string> &aggregate_sql,
 
95
                    std::vector<std::string> &segmented_sql);
 
96
 
 
97
  /**
 
98
   * Execute a batch of SQL statements.
 
99
   *
 
100
   * @param sql Batch of SQL statements to execute.
 
101
   * @param commit_id Commit ID value to store in state table.
 
102
   *
 
103
   * @retval true Success
 
104
   * @retval false Failure
 
105
   */
 
106
  bool executeSQLWithCommitId(std::vector<std::string> &sql,
 
107
                              const std::string &commit_id);
 
108
  
 
109
  /**
 
110
   * Remove messages for a given transaction from the queue.
 
111
   *
 
112
   * @param trx_id Transaction ID for the messages to remove.
 
113
   *
 
114
   * @retval true Success
 
115
   * @retval false Failure
 
116
   */
 
117
  bool deleteFromQueue(uint64_t trx_id);
 
118
 
 
119
  /**
 
120
   * Determine if a Statement message is an end message.
 
121
   *
 
122
   * @retval true Is an end Statement message
 
123
   * @retval false Is NOT an end Statement message
 
124
   */
 
125
  bool isEndStatement(const drizzled::message::Statement &statement);
 
126
};
 
127
 
 
128
} /* namespace slave */
 
129
 
 
130
#endif /* PLUGIN_SLAVE_QUEUE_CONSUMER_H */