1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
|
/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-
* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
*
* Copyright (C) 2008-2009 Sun Microsystems
*
* Authors:
*
* Jay Pipes <joinfu@sun.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; version 2 of the License.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*/
#ifndef DRIZZLED_REPLICATION_SERVICES_H
#define DRIZZLED_REPLICATION_SERVICES_H
#include "drizzled/atomics.h"
#include "drizzled/message/transaction.pb.h"
#include <vector>
/* some forward declarations needed */
class Session;
class Table;
namespace drizzled
{
namespace plugin
{
class TransactionReplicator;
class TransactionApplier;
}
/**
* This is a class which manages transforming internal
* transactional events into GPB messages and sending those
* events out through registered replicators and appliers.
*/
class ReplicationServices
{
public:
static const size_t DEFAULT_RECORD_SIZE= 100;
typedef uint64_t GlobalTransactionId;
/**
* Types of messages that can go in the transaction
* log file. Every time something is written into the
* transaction log, it is preceded by a header containing
* the type of message which follows.
*/
enum MessageType
{
TRANSACTION= 1, /* A GPB Transaction Message */
BLOB= 2 /* A BLOB value */
};
typedef std::vector<plugin::TransactionReplicator *> Replicators;
typedef std::vector<plugin::TransactionApplier *> Appliers;
private:
/**
* Atomic boolean set to true if any *active* replicators
* or appliers are actually registered.
*/
atomic<bool> is_active;
/**
* The timestamp of the last time a Transaction message was successfully
* applied (sent to an Applier)
*/
atomic<uint64_t> last_applied_timestamp;
/** Our collection of replicator plugins */
Replicators replicators;
/** Our collection of applier plugins */
Appliers appliers;
/**
* Helper method which is called after any change in the
* registered appliers or replicators to evaluate whether
* any remaining plugins are actually active.
*
* This method properly sets the is_active member variable.
*/
void evaluateActivePlugins();
/**
* Helper method which returns the active Transaction message
* for the supplied Session. If one is not found, a new Transaction
* message is allocated, initialized, and returned.
*
* @param The session processing the transaction
*/
drizzled::message::Transaction *getActiveTransaction(Session *in_session) const;
/**
* Helper method which attaches a transaction context
* the supplied transaction based on the supplied Session's
* transaction information. This method also ensure the
* transaction message is attached properly to the Session object
*
* @param The transaction message to initialize
* @param The Session processing this transaction
*/
void initTransaction(drizzled::message::Transaction &in_command, Session *in_session) const;
/**
* Helper method which finalizes data members for the
* supplied transaction's context.
*
* @param The transaction message to finalize
* @param The Session processing this transaction
*/
void finalizeTransaction(drizzled::message::Transaction &in_command, Session *in_session) const;
/**
* Helper method which deletes transaction memory and
* unsets Session's transaction and statement messages.
*/
void cleanupTransaction(message::Transaction *in_transaction,
Session *in_session) const;
/**
* Returns true if the transaction contains any Statement
* messages which are not end segments (i.e. a bulk statement has
* previously been sent to replicators).
*
* @param The transaction to check
*/
bool transactionContainsBulkSegment(const drizzled::message::Transaction &transaction) const;
/**
* Helper method which initializes a Statement message
*
* @param The statement to initialize
* @param The type of the statement
* @param The session processing this statement
*/
void initStatement(drizzled::message::Statement &statement,
drizzled::message::Statement::Type in_type,
Session *in_session) const;
/**
* Helper method which returns an initialized Statement
* message for methods doing insertion of data.
*
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table object being inserted into
*/
message::Statement &getInsertStatement(Session *in_session,
Table *in_table) const;
/**
* Helper method which initializes the header message for
* insert operations.
*
* @param[inout] Statement message container to modify
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table being inserted into
*/
void setInsertHeader(message::Statement &statement,
Session *in_session,
Table *in_table) const;
/**
* Helper method which returns an initialized Statement
* message for methods doing updates of data.
*
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table object being updated
* @param[in] Pointer to the old data in the record
* @param[in] Pointer to the new data in the record
*/
message::Statement &getUpdateStatement(Session *in_session,
Table *in_table,
const unsigned char *old_record,
const unsigned char *new_record) const;
/**
* Helper method which initializes the header message for
* update operations.
*
* @param[inout] Statement message container to modify
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table being updated
* @param[in] Pointer to the old data in the record
* @param[in] Pointer to the new data in the record
*/
void setUpdateHeader(message::Statement &statement,
Session *in_session,
Table *in_table,
const unsigned char *old_record,
const unsigned char *new_record) const;
/**
* Helper method which returns an initialized Statement
* message for methods doing deletion of data.
*
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table object being deleted from
*/
message::Statement &getDeleteStatement(Session *in_session,
Table *in_table) const;
/**
* Helper method which initializes the header message for
* insert operations.
*
* @param[inout] Statement message container to modify
* @param[in] Pointer to the Session doing the processing
* @param[in] Pointer to the Table being deleted from
*/
void setDeleteHeader(message::Statement &statement,
Session *in_session,
Table *in_table) const;
/**
* Helper method which pushes a constructed message out
* to the registered replicator and applier plugins.
*
* @param Message to push out
*/
void push(drizzled::message::Transaction &to_push);
public:
/**
* Constructor
*/
ReplicationServices();
/**
* Singleton method
* Returns the singleton instance of ReplicationServices
*/
static inline ReplicationServices &singleton()
{
static ReplicationServices replication_services;
return replication_services;
}
/**
* Returns whether the ReplicationServices object
* is active. In other words, does it have both
* a replicator and an applier that are *active*?
*/
bool isActive() const;
/**
* Attaches a replicator to our internal collection of
* replicators.
*
* @param Pointer to a replicator to attach/register
*/
void attachReplicator(drizzled::plugin::TransactionReplicator *in_replicator);
/**
* Detaches/unregisters a replicator with our internal
* collection of replicators.
*
* @param Pointer to the replicator to detach
*/
void detachReplicator(drizzled::plugin::TransactionReplicator *in_replicator);
/**
* Attaches a applier to our internal collection of
* appliers.
*
* @param Pointer to a applier to attach/register
*/
void attachApplier(drizzled::plugin::TransactionApplier *in_applier);
/**
* Detaches/unregisters a applier with our internal
* collection of appliers.
*
* @param Pointer to the applier to detach
*/
void detachApplier(drizzled::plugin::TransactionApplier *in_applier);
/**
* Commits a normal transaction (see above) and pushes the
* transaction message out to the replicators.
*
* @param Pointer to the Session committing the transaction
*/
void commitTransaction(Session *in_session);
/**
* Marks the current active transaction message as being rolled
* back and pushes the transaction message out to replicators.
*
* @param Pointer to the Session committing the transaction
*/
void rollbackTransaction(Session *in_session);
/**
* Finalizes a Statement message and sets the Session's statement
* message to NULL.
*
* @param The statement to initialize
* @param The session processing this statement
*/
void finalizeStatement(drizzled::message::Statement &statement,
Session *in_session) const;
/**
* Creates a new InsertRecord GPB message and pushes it to
* replicators.
*
* @param Pointer to the Session which has inserted a record
* @param Pointer to the Table containing insert information
*
* Grr, returning "true" here on error because of the cursor
* reversed bool return crap...fix that.
*/
bool insertRecord(Session *in_session, Table *in_table);
/**
* Creates a new UpdateRecord GPB message and pushes it to
* replicators.
*
* @param Pointer to the Session which has updated a record
* @param Pointer to the Table containing update information
* @param Pointer to the raw bytes representing the old record/row
* @param Pointer to the raw bytes representing the new record/row
*/
void updateRecord(Session *in_session,
Table *in_table,
const unsigned char *old_record,
const unsigned char *new_record);
/**
* Creates a new DeleteRecord GPB message and pushes it to
* replicators.
*
* @param Pointer to the Session which has deleted a record
* @param Pointer to the Table containing delete information
*/
void deleteRecord(Session *in_session, Table *in_table);
/**
* Creates a TruncateTable Statement GPB message and add it
* to the Session's active Transaction GPB message for pushing
* out to the replicator streams.
*
* @param[in] Pointer to the Session which issued the statement
* @param[in] The Table being truncated
*/
void truncateTable(Session *in_session, Table *in_table);
/**
* Creates a new RawSql GPB message and pushes it to
* replicators.
*
* @TODO With a real data dictionary, this really shouldn't
* be needed. CREATE TABLE would map to insertRecord call
* on the I_S, etc. Not sure what to do with administrative
* commands like CHECK TABLE, though..
*
* @param Pointer to the Session which issued the statement
* @param Query string
* @param Length of the query string
*/
void rawStatement(Session *in_session, const char *in_query, size_t in_query_len);
/**
* Returns the timestamp of the last Transaction which was sent to
* an applier.
*/
uint64_t getLastAppliedTimestamp() const;
};
} /* end namespace drizzled */
#endif /* DRIZZLED_REPLICATION_SERVICES_H */
|