Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 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 | 428x 428x 428x 428x 428x 428x 428x 428x 428x 428x 428x 428x 428x 233x 233x 233x 233x 233x 428x 712x 712x 600x 600x 108x 108x 108x 4x 4x 4x 428x 346x 346x 268x 268x 74x 74x 74x 4x 182x 14x 428x 428x 428x 19300805x 19300805x 19300471x 19300471x 19300471x 334x 334x 334x 334x 334x 334x 334x | import { promisify } from 'util';
import * as zlib from 'zlib';
import { LEGACY_HELLO_COMMAND } from '../../constants';
import { getSnappy, getZstdLibrary, type SnappyLib, type ZStandard } from '../../deps';
import { MongoDecompressionError, MongoInvalidArgumentError } from '../../error';
import {
type MessageHeader,
OpCompressedRequest,
OpMsgResponse,
OpReply,
type WriteProtocolMessageType
} from '../commands';
import { OP_COMPRESSED, OP_MSG } from './constants';
/** @public */
export const Compressor = Object.freeze({
none: 0,
snappy: 1,
zlib: 2,
zstd: 3
} as const);
/** @public */
export type Compressor = (typeof Compressor)[CompressorName];
/** @public */
export type CompressorName = keyof typeof Compressor;
export const uncompressibleCommands = new Set([
LEGACY_HELLO_COMMAND,
'saslStart',
'saslContinue',
'getnonce',
'authenticate',
'createUser',
'updateUser',
'copydbSaslStart',
'copydbgetnonce',
'copydb'
]);
const ZSTD_COMPRESSION_LEVEL = 3;
const zlibInflate = promisify(zlib.inflate.bind(zlib));
const zlibDeflate = promisify(zlib.deflate.bind(zlib));
let zstd: ZStandard;
let Snappy: SnappyLib | null = null;
function loadSnappy() {
Eif (Snappy == null) {
const snappyImport = getSnappy();
Iif ('kModuleError' in snappyImport) {
throw snappyImport.kModuleError;
}
Snappy = snappyImport;
}
return Snappy;
}
// Facilitate compressing a message using an agreed compressor
export async function compress(
options: { zlibCompressionLevel: number; agreedCompressor: CompressorName },
dataToBeCompressed: Buffer
): Promise<Buffer> {
const zlibOptions = {} as zlib.ZlibOptions;
switch (options.agreedCompressor) {
case 'snappy': {
Snappy ??= loadSnappy();
return await Snappy.compress(dataToBeCompressed);
}
case 'zstd': {
loadZstd();
Iif ('kModuleError' in zstd) {
throw zstd['kModuleError'];
}
return await zstd.compress(dataToBeCompressed, ZSTD_COMPRESSION_LEVEL);
}
case 'zlib': {
Eif (options.zlibCompressionLevel) {
zlibOptions.level = options.zlibCompressionLevel;
}
return await zlibDeflate(dataToBeCompressed, zlibOptions);
}
default: {
throw new MongoInvalidArgumentError(
`Unknown compressor ${options.agreedCompressor} failed to compress`
);
}
}
}
// Decompress a message using the given compressor
export async function decompress(compressorID: number, compressedData: Buffer): Promise<Buffer> {
Iif (
compressorID !== Compressor.snappy &&
compressorID !== Compressor.zstd &&
compressorID !== Compressor.zlib &&
compressorID !== Compressor.none
) {
throw new MongoDecompressionError(
`Server sent message compressed using an unsupported compressor. (Received compressor ID ${compressorID})`
);
}
switch (compressorID) {
case Compressor.snappy: {
Snappy ??= loadSnappy();
return await Snappy.uncompress(compressedData, { asBuffer: true });
}
case Compressor.zstd: {
loadZstd();
Iif ('kModuleError' in zstd) {
throw zstd['kModuleError'];
}
return await zstd.decompress(compressedData);
}
case Compressor.zlib: {
return await zlibInflate(compressedData);
}
default: {
return compressedData;
}
}
}
/**
* Load ZStandard if it is not already set.
*/
function loadZstd() {
if (!zstd) {
zstd = getZstdLibrary();
}
}
const MESSAGE_HEADER_SIZE = 16;
/**
* @internal
*
* Compresses an OP_MSG or OP_QUERY message, if compression is configured. This method
* also serializes the command to BSON.
*/
export async function compressCommand(
command: WriteProtocolMessageType,
description: { agreedCompressor?: CompressorName; zlibCompressionLevel?: number }
): Promise<Buffer> {
const finalCommand =
description.agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)
? command
: new OpCompressedRequest(command, {
agreedCompressor: description.agreedCompressor ?? 'none',
zlibCompressionLevel: description.zlibCompressionLevel ?? 0
});
const data = await finalCommand.toBin();
return Buffer.concat(data);
}
/**
* @internal
*
* Decompresses an OP_MSG or OP_QUERY response from the server, if compression is configured.
*
* This method does not parse the response's BSON.
*/
export async function decompressResponse(message: Buffer): Promise<OpMsgResponse | OpReply> {
const messageHeader: MessageHeader = {
length: message.readInt32LE(0),
requestId: message.readInt32LE(4),
responseTo: message.readInt32LE(8),
opCode: message.readInt32LE(12)
};
if (messageHeader.opCode !== OP_COMPRESSED) {
const ResponseType = messageHeader.opCode === OP_MSG ? OpMsgResponse : OpReply;
const messageBody = message.subarray(MESSAGE_HEADER_SIZE);
return new ResponseType(message, messageHeader, messageBody);
}
const header: MessageHeader = {
...messageHeader,
fromCompressed: true,
opCode: message.readInt32LE(MESSAGE_HEADER_SIZE),
length: message.readInt32LE(MESSAGE_HEADER_SIZE + 4)
};
const compressorID = message[MESSAGE_HEADER_SIZE + 8];
const compressedBuffer = message.slice(MESSAGE_HEADER_SIZE + 9);
// recalculate based on wrapped opcode
const ResponseType = header.opCode === OP_MSG ? OpMsgResponse : OpReply;
const messageBody = await decompress(compressorID, compressedBuffer);
Iif (messageBody.length !== header.length) {
throw new MongoDecompressionError('Message body and message header must be the same length');
}
return new ResponseType(message, header, messageBody);
}
|