msg.js
6.54 KB
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
'use strict';
// Implementation of OP_MSG spec:
// https://github.com/mongodb/specifications/blob/master/source/message/OP_MSG.rst
//
// struct Section {
// uint8 payloadType;
// union payload {
// document document; // payloadType == 0
// struct sequence { // payloadType == 1
// int32 size;
// cstring identifier;
// document* documents;
// };
// };
// };
// struct OP_MSG {
// struct MsgHeader {
// int32 messageLength;
// int32 requestID;
// int32 responseTo;
// int32 opCode = 2013;
// };
// uint32 flagBits;
// Section+ sections;
// [uint32 checksum;]
// };
const Buffer = require('safe-buffer').Buffer;
const opcodes = require('../wireprotocol/shared').opcodes;
const databaseNamespace = require('../wireprotocol/shared').databaseNamespace;
const ReadPreference = require('../topologies/read_preference');
// Incrementing request id
let _requestId = 0;
// Msg Flags
const OPTS_CHECKSUM_PRESENT = 1;
const OPTS_MORE_TO_COME = 2;
const OPTS_EXHAUST_ALLOWED = 1 << 16;
class Msg {
constructor(bson, ns, command, options) {
// Basic options needed to be passed in
if (command == null) throw new Error('query must be specified for query');
// Basic options
this.bson = bson;
this.ns = ns;
this.command = command;
this.command.$db = databaseNamespace(ns);
if (options.readPreference && options.readPreference.mode !== ReadPreference.PRIMARY) {
this.command.$readPreference = options.readPreference.toJSON();
}
// Ensure empty options
this.options = options || {};
// Additional options
this.requestId = options.requestId ? options.requestId : Msg.getRequestId();
// Serialization option
this.serializeFunctions =
typeof options.serializeFunctions === 'boolean' ? options.serializeFunctions : false;
this.ignoreUndefined =
typeof options.ignoreUndefined === 'boolean' ? options.ignoreUndefined : false;
this.checkKeys = typeof options.checkKeys === 'boolean' ? options.checkKeys : false;
this.maxBsonSize = options.maxBsonSize || 1024 * 1024 * 16;
// flags
this.checksumPresent = false;
this.moreToCome = options.moreToCome || false;
this.exhaustAllowed =
typeof options.exhaustAllowed === 'boolean' ? options.exhaustAllowed : false;
}
toBin() {
const buffers = [];
let flags = 0;
if (this.checksumPresent) {
flags |= OPTS_CHECKSUM_PRESENT;
}
if (this.moreToCome) {
flags |= OPTS_MORE_TO_COME;
}
if (this.exhaustAllowed) {
flags |= OPTS_EXHAUST_ALLOWED;
}
const header = Buffer.alloc(
4 * 4 + // Header
4 // Flags
);
buffers.push(header);
let totalLength = header.length;
const command = this.command;
totalLength += this.makeDocumentSegment(buffers, command);
header.writeInt32LE(totalLength, 0); // messageLength
header.writeInt32LE(this.requestId, 4); // requestID
header.writeInt32LE(0, 8); // responseTo
header.writeInt32LE(opcodes.OP_MSG, 12); // opCode
header.writeUInt32LE(flags, 16); // flags
return buffers;
}
makeDocumentSegment(buffers, document) {
const payloadTypeBuffer = Buffer.alloc(1);
payloadTypeBuffer[0] = 0;
const documentBuffer = this.serializeBson(document);
buffers.push(payloadTypeBuffer);
buffers.push(documentBuffer);
return payloadTypeBuffer.length + documentBuffer.length;
}
serializeBson(document) {
return this.bson.serialize(document, {
checkKeys: this.checkKeys,
serializeFunctions: this.serializeFunctions,
ignoreUndefined: this.ignoreUndefined
});
}
}
Msg.getRequestId = function() {
_requestId = (_requestId + 1) & 0x7fffffff;
return _requestId;
};
class BinMsg {
constructor(bson, message, msgHeader, msgBody, opts) {
opts = opts || { promoteLongs: true, promoteValues: true, promoteBuffers: false };
this.parsed = false;
this.raw = message;
this.data = msgBody;
this.bson = bson;
this.opts = opts;
// Read the message header
this.length = msgHeader.length;
this.requestId = msgHeader.requestId;
this.responseTo = msgHeader.responseTo;
this.opCode = msgHeader.opCode;
this.fromCompressed = msgHeader.fromCompressed;
// Read response flags
this.responseFlags = msgBody.readInt32LE(0);
this.checksumPresent = (this.responseFlags & OPTS_CHECKSUM_PRESENT) !== 0;
this.moreToCome = (this.responseFlags & OPTS_MORE_TO_COME) !== 0;
this.exhaustAllowed = (this.responseFlags & OPTS_EXHAUST_ALLOWED) !== 0;
this.promoteLongs = typeof opts.promoteLongs === 'boolean' ? opts.promoteLongs : true;
this.promoteValues = typeof opts.promoteValues === 'boolean' ? opts.promoteValues : true;
this.promoteBuffers = typeof opts.promoteBuffers === 'boolean' ? opts.promoteBuffers : false;
this.documents = [];
}
isParsed() {
return this.parsed;
}
parse(options) {
// Don't parse again if not needed
if (this.parsed) return;
options = options || {};
this.index = 4;
// Allow the return of raw documents instead of parsing
const raw = options.raw || false;
const documentsReturnedIn = options.documentsReturnedIn || null;
const promoteLongs =
typeof options.promoteLongs === 'boolean' ? options.promoteLongs : this.opts.promoteLongs;
const promoteValues =
typeof options.promoteValues === 'boolean' ? options.promoteValues : this.opts.promoteValues;
const promoteBuffers =
typeof options.promoteBuffers === 'boolean'
? options.promoteBuffers
: this.opts.promoteBuffers;
// Set up the options
const _options = {
promoteLongs: promoteLongs,
promoteValues: promoteValues,
promoteBuffers: promoteBuffers
};
while (this.index < this.data.length) {
const payloadType = this.data.readUInt8(this.index++);
if (payloadType === 1) {
console.error('TYPE 1');
} else if (payloadType === 0) {
const bsonSize = this.data.readUInt32LE(this.index);
const bin = this.data.slice(this.index, this.index + bsonSize);
this.documents.push(raw ? bin : this.bson.deserialize(bin, _options));
this.index += bsonSize;
}
}
if (this.documents.length === 1 && documentsReturnedIn != null && raw) {
const fieldsAsRaw = {};
fieldsAsRaw[documentsReturnedIn] = true;
_options.fieldsAsRaw = fieldsAsRaw;
const doc = this.bson.deserialize(this.documents[0], _options);
this.documents = [doc];
}
this.parsed = true;
}
}
module.exports = { Msg, BinMsg };