← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/serialize/binary_log.cc

  • Committer: Mats Kindahl
  • Date: 2008-08-19 07:43:02 UTC
  • mto: This revision was merged to the branch mainline in revision 350.
  • Revision ID: mats@mysql.com-20080819074302-78dxfv9ukmo3jj3d
Adding specification of a simple protobuf-based binary log format,
including a reader and a writer. The intention is to produce some
tools to experiment with the format and see what needs to be done.

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/serialize/binary_log.cc
 
1
#include "binary_log.h"
 
2
 
 
3
#include <google/protobuf/io/coded_stream.h>
 
4
 
 
5
using namespace google::protobuf;
 
6
using namespace google::protobuf::io;
 
7
 
 
8
bool
 
9
BinaryLog::Event::write(CodedOutputStream* out) const
 
10
{
 
11
  // We frame each event in a length encoded in a special manner, and
 
12
  // end it with a CRC-32 checksum.
 
13
 
 
14
  // Write length and type
 
15
  unsigned char buf[LENGTH_ENCODE_MAX_BYTES + 1];
 
16
  unsigned char *end= length_encode(m_message->ByteSize(), buf);
 
17
  *end++= m_type;
 
18
 
 
19
  char cs[4] = { 0 };                           // !!! No checksum yet
 
20
  if (!out->WriteRaw(buf, end - buf) ||         // Length + Type
 
21
      !m_message->SerializeToCodedStream(out) || // Event body
 
22
      !out->WriteRaw(cs, sizeof(cs)))           // Checksum
 
23
    return false;
 
24
 
 
25
  return true;
 
26
}
 
27
 
 
28
 
 
29
bool
 
30
BinaryLog::Event::read(CodedInputStream *in)
 
31
{
 
32
  unsigned char buf[LENGTH_ENCODE_MAX_BYTES + 1];
 
33
 
 
34
  // Read length peek byte to figure out length
 
35
  if (!in->ReadRaw(buf, 1))
 
36
    return false;
 
37
 
 
38
  // Read in the rest of the length bytes plus the type
 
39
  size_t bytes= length_decode_bytes(*buf);
 
40
  if (!in->ReadRaw(buf + 1, bytes))
 
41
    return false;
 
42
 
 
43
  size_t length;
 
44
  (void) length_decode(buf, &length);
 
45
 
 
46
  // Fetch type from read buffer
 
47
  m_type= static_cast<EventType>(buf[bytes]);
 
48
 
 
49
  // Create the right event based on the type code (is there something
 
50
  // better in the protobuf library?)
 
51
  Message *message= NULL;
 
52
  switch (m_type) {
 
53
  case QUERY:
 
54
    message= new BinaryLog::Query;
 
55
    break;
 
56
 
 
57
  case COMMIT:
 
58
    message= new BinaryLog::Commit;
 
59
    break;
 
60
 
 
61
  case ROLLBACK:
 
62
    message= new BinaryLog::Rollback;
 
63
    break;
 
64
 
 
65
  case START:
 
66
    message= new BinaryLog::Start;
 
67
    break;
 
68
 
 
69
  case CHAIN:
 
70
    message= new BinaryLog::Chain;
 
71
    break;
 
72
  }
 
73
 
 
74
  if (!message)
 
75
    return false;
 
76
 
 
77
  // Read the event body as length bytes. It is necessary to limit the
 
78
  // stream since otherwise ParseFromCodedStream reads all bytes of
 
79
  // the stream.
 
80
  CodedInputStream::Limit limit= in->PushLimit(length);
 
81
  if (!message->ParseFromCodedStream(in))
 
82
    return false;
 
83
  in->PopLimit(limit);
 
84
  delete m_message;
 
85
  m_message= message;
 
86
 
 
87
  // Read checksum (none here yet)
 
88
  char checksum[4];
 
89
  if (!in->ReadRaw(checksum, sizeof(checksum)))
 
90
    return false;
 
91
  return true;
 
92
}
 
93
 
 
94
template <class EventClass>
 
95
void print_common(std::ostream& out, EventClass* event)
 
96
{
 
97
  out << "# Global Id: (" << event->header().server_id() << "," << event->header().trans_id() << ")\n";
 
98
}
 
99
 
 
100
 
 
101
void
 
102
BinaryLog::Event::print(std::ostream& out) const
 
103
{
 
104
  using namespace google::protobuf;
 
105
 
 
106
  switch (m_type) {
 
107
  case QUERY:
 
108
  {
 
109
    Query *event= static_cast<Query*>(m_message);
 
110
    print_common(out, event);
 
111
    for (RepeatedPtrField<Query::Variable>::const_iterator ii= event->variable().begin() ;
 
112
         ii != event->variable().end() ;
 
113
         ++ii)
 
114
    {
 
115
      out << "set @" << ii->name() << " = '" << ii->value() << "'\n";
 
116
    }
 
117
    out << event->query() << std::endl;
 
118
    break;
 
119
  }
 
120
 
 
121
  case COMMIT:
 
122
  {
 
123
    Commit *event= static_cast<Commit*>(m_message);
 
124
    print_common(out, event);
 
125
    // NYI !!!
 
126
    break;
 
127
  }
 
128
 
 
129
  case ROLLBACK:
 
130
  {
 
131
    Rollback *event= static_cast<Rollback*>(m_message);
 
132
    print_common(out, event);
 
133
    // NYI !!!
 
134
    break;
 
135
  }
 
136
 
 
137
  case START:
 
138
  {
 
139
    Start *event= static_cast<Start*>(m_message);
 
140
    print_common(out, event);
 
141
    // NYI !!!
 
142
    break;
 
143
  }
 
144
 
 
145
  case CHAIN:
 
146
  {
 
147
    Chain *event= static_cast<Chain*>(m_message);
 
148
    print_common(out, event);
 
149
    // NYI !!!
 
150
    break;
 
151
  }
 
152
 
 
153
  default:
 
154
    break;                                      /* Nothing */
 
155
  }
 
156
}