neton.wal.Encoder source code

1 module neton.wal.Encoder;
2 
3 import std.file;
4 import neton.wal.Record;
5 import hunt.util.Serialize;
6 import std.stdio;
7 import hunt.logging;
8 // walPageBytes is the alignment for flushing records to the backing Writer.
9 // It should be a multiple of the minimum sector size so that WAL can safely
10 // distinguish between torn writes and ordinary data corruption.
11 //const walPageBytes = 8 * minSectorSize;
12 
13 /* encode form 
14 
15  | 8 bit |   56 bit    | data | ...... | 8 bit |   56 bit    | data | .......
16      |         |
17  | type  | data size   |               | type  | data size   |
18  -----------------------               -----------------------
19           ||                                      ||
20        uint64(8 byte)                       uint64(8 byte)
21 
22 */
23 
24 class Encoder  {
25 
26 	byte[]    _buf;
27     File      _fw;
28 
29 
30     this(File fp) {
31         _fw = fp;
32        // _buf = new byte[1024*1024];
33     }
34 
35 
36     void  encode(Record rec)  {
37         
38         //_fw.seek(0,SEEK_END);
39         byte[] data;
40 
41         data = serialize(rec);
42         //logInfo("--------------debug 3-----",data," file name : ",_fw.name);
43         ulong lenField= encodeFrameSize(data.length,WalType.recordType);
44         //logInfo("--------------debug 4-----",lenField," data.length : ",data.length);
45         writeUint64(lenField);
46         //logInfo("--------------debug 5-----"," file name : ",_fw.name);
47         _fw.rawWrite(data);
48         return;
49     }
50 
51     ulong encodeFrameSize(ulong dataBytes, WalType type) {
52         ulong t = cast(ulong)type;
53         return (cast(ulong)(t << 56) | dataBytes);
54     }
55 
56     void flush()  {
57         //logInfo("----- flush file : ",_fw.name," file open : ",_fw.isOpen);
58         if(_fw.isOpen)
59             _fw.flush();
60     }
61 
62     void  writeUint64(ulong n)  {
63         byte[] uint64buf = new byte[8];
64         for(int i = 0 ; i< 8 ; i++ )
65             uint64buf[i] =cast(byte)( (cast(byte)(n >>(8*(8-i-1)))) & 0xff);
66         //logInfo("--------------debug 6-----",uint64buf," file name : ",_fw.name, " file open : ",_fw.isOpen);
67         try
68         {
69             if(!_fw.isOpen)
70                 _fw.open(_fw.name,"ab+");
71             _fw.rawWrite(uint64buf);
72             // logInfo("--------------debug 7-----");
73         }
74         catch (Exception e)
75         {
76             logError("-----7 catch :", e.msg);
77         }
78 
79         return;
80     }
81 
82     ulong curOff()
83     {
84         return _fw.tell();
85     }
86 }
87