Revision f12fdfae

View differences:

include/LogDB.h
65 65
    /**
66 66
     *  SQL callback to load logDBRecord from DB (SELECT commands)
67 67
     */
68
    int select_cb(void *nil, int num, char **values, char **names)
69
    {
70
        if ( !values || !values[0] || !values[1] || !values[2] || !values[3] ||
71
                !values[4] || !values[5] || num != 6 )
72
        {
73
            return -1;
74
        }
75

  
76
        index = static_cast<unsigned int>(atoi(values[0]));
77
        term  = static_cast<unsigned int>(atoi(values[1]));
78
        sql   = values[2];
79

  
80
        timestamp  = static_cast<unsigned int>(atoi(values[3]));
81

  
82
        prev_index = static_cast<unsigned int>(atoi(values[4]));
83
        prev_term  = static_cast<unsigned int>(atoi(values[5]));
84

  
85
        return 0;
86
    }
68
    int select_cb(void *nil, int num, char **values, char **names);
87 69
};
88 70

  
89 71
/**
......
112 94
    /**
113 95
     *  Applies the SQL command of the given record to the database. The
114 96
     *  timestamp of the record is updated.
115
     *    @param lr the log record
116 97
     *    @param index of the log record
117 98
     */
118
    int apply_log_record(LogDBRecord * lr);
119

  
120 99
	int apply_log_records(unsigned int commit_index);
121 100

  
122 101
    /**
......
279 258
    static const char * db_bootstrap;
280 259

  
281 260
    /**
261
     *  Applies the SQL command of the given record to the database. The
262
     *  timestamp of the record is updated.
263
     *    @param lr the log record
264
     */
265
    int apply_log_record(LogDBRecord * lr);
266

  
267
    /**
282 268
     *  Inserts or update a log record in the database
283 269
     *    @param index of the log entry
284 270
     *    @param term for the log entry
src/raft/RaftManager.cc
370 370

  
371 371
    requests.clear();
372 372

  
373
    leader_hook->do_hook(0);
373
    if ( leader_hook != 0 )
374
    {
375
        leader_hook->do_hook(0);
376
    }
374 377

  
375 378
    state = LEADER;
376 379

  
......
431 434

  
432 435
    pthread_mutex_lock(&mutex);
433 436

  
434
    if ( state == LEADER )
437
    if ( state == LEADER && follower_hook != 0 )
435 438
    {
436 439
        follower_hook->do_hook(0);
437 440
    }
src/sql/LogDB.cc
16 16

  
17 17
#include "LogDB.h"
18 18
#include "Nebula.h"
19
#include "NebulaUtil.h"
19 20
#include "ZoneServer.h"
20 21
#include "Callbackable.h"
21 22

  
......
33 34
/* -------------------------------------------------------------------------- */
34 35
/* -------------------------------------------------------------------------- */
35 36

  
37
int LogDBRecord::select_cb(void *nil, int num, char **values, char **names)
38
{
39
    if ( !values || !values[0] || !values[1] || !values[2] || !values[3] ||
40
            !values[4] || !values[5] || num != 6 )
41
    {
42
        return -1;
43
    }
44

  
45
    std::string zsql;
46

  
47
    std::string * _sql;
48

  
49
    index = static_cast<unsigned int>(atoi(values[0]));
50
    term  = static_cast<unsigned int>(atoi(values[1]));
51
    zsql  = values[2];
52

  
53
    timestamp  = static_cast<unsigned int>(atoi(values[3]));
54

  
55
    prev_index = static_cast<unsigned int>(atoi(values[4]));
56
    prev_term  = static_cast<unsigned int>(atoi(values[5]));
57

  
58
    _sql = one_util::zlib_decompress(zsql, true);
59

  
60
    if ( _sql == 0 )
61
    {
62
        return -1;
63
    }
64

  
65
    sql = *_sql;
66

  
67
    delete _sql;
68

  
69
    return 0;
70
}
71

  
72
/* -------------------------------------------------------------------------- */
73
/* -------------------------------------------------------------------------- */
74

  
36 75
LogDB::LogDB(SqlDB * _db, bool _solo, unsigned int _lret):solo(_solo), db(_db),
37 76
    next_index(0), last_applied(-1), last_index(-1), last_term(-1),
38 77
    log_retention(_lret)
......
155 194
{
156 195
    ostringstream oss;
157 196

  
197
    std::string zraft_xml;
198

  
158 199
    single_cb<std::string> cb;
159 200

  
160 201
    oss << "SELECT sql FROM logdb WHERE log_index = -1 AND term = -1";
161 202

  
162
    cb.set_callback(&raft_xml);
203
    cb.set_callback(&zraft_xml);
163 204

  
164 205
    int rc = db->exec_rd(oss, &cb);
165 206

  
166 207
    cb.unset_callback();
167 208

  
168
    if ( raft_xml.empty() )
209
    if ( zraft_xml.empty() )
169 210
    {
170 211
        rc = -1;
171 212
    }
172 213

  
214
    std::string * _raft_xml = one_util::zlib_decompress(zraft_xml, true);
215

  
216
    if ( _raft_xml == 0 )
217
    {
218
        return -1;
219
    }
220

  
221
    raft_xml = *_raft_xml;
222

  
223
    delete _raft_xml;
224

  
173 225
    return rc;
174 226
}
175 227

  
......
181 233
{
182 234
    std::ostringstream oss;
183 235

  
184
    char * sql_db = db->escape_str(sql.c_str());
236
    std::string * zsql;
237

  
238
    zsql = one_util::zlib_compress(sql, true);
239

  
240
    if ( zsql == 0 )
241
    {
242
        return -1;
243
    }
244

  
245
    char * sql_db = db->escape_str(zsql->c_str());
246

  
247
    delete zsql;
185 248

  
186 249
    if ( sql_db == 0 )
187 250
    {
......
360 423

  
361 424
    pthread_mutex_lock(&mutex);
362 425

  
363
    oss << "DELETE FROM " << table << " WHERE log_index >= start_index";
426
    oss << "DELETE FROM " << table << " WHERE log_index >= " << start_index;
364 427

  
365 428
    rc = db->exec_wr(oss);
366 429

  

Also available in: Unified diff