Merge pull request #447 from ml054/RDBC-858

RDBC-858 unify http compression/decompression API

Author: ml054

src/Documents/BulkInsert/BulkInsertWriterBase.ts

@@ -4,6 +4,7 @@ import { pipeline, Readable } from "node:stream";
 import type { Gzip } from "node:zlib";
 import { promisify } from "node:util";
 import { TypeUtil } from "../../Utility/TypeUtil.js";
+import { HttpCompressionAlgorithm } from "../../Http/HttpCompressionAlgorithm.js";
 
 export class BulkInsertWriterBase implements IDisposable {
     private readonly _maxSizeInBuffer = 1024 * 1024;
@@ -97,8 +98,8 @@ export class BulkInsertWriterBase implements IDisposable {
         }
     }
 
-    public async ensureStream(compression: boolean) {
-        if (compression) {
+    public async ensureStream(compression: HttpCompressionAlgorithm) {
+        if (compression === "Gzip") {
             const { createGzip } = await import("node:zlib");
             this.compressedStream = createGzip();
             pipeline(this.requestBodyStream, this.compressedStream, TypeUtil.NOOP);

src/Documents/BulkInsertOperation.ts

@@ -34,6 +34,7 @@ import { Semaphore } from "../Utility/Semaphore.js";
 import { BulkInsertOperationBase } from "./BulkInsert/BulkInsertOperationBase.js";
 import { BulkInsertOptions } from "./BulkInsert/BulkInsertOptions.js";
 import { BulkInsertWriter } from "./BulkInsert/BulkInsertWriter.js";
+import { HttpCompressionAlgorithm } from "../Http/HttpCompressionAlgorithm.js";
 
 export class BulkInsertOperation extends BulkInsertOperationBase<object> {
 
@@ -750,11 +751,11 @@ export class BulkInsertOperation extends BulkInsertOperationBase<object> {
     }
 
     protected async _ensureStream() {
-        await this._writer.ensureStream(this.useCompression);
+        const compressionAlgorithm = this._useCompression ? this._conventions.httpCompressionAlgorithm : null;
+        await this._writer.ensureStream(compressionAlgorithm);
 
         const bulkCommand =
-            new BulkInsertCommand(this._operationId, this._writer.compressedStream ?? this._writer.requestBodyStream, this._nodeTag, this._options.skipOverwriteIfUnchanged);
-        bulkCommand.useCompression = this._useCompression;
+            new BulkInsertCommand(this._operationId, compressionAlgorithm, this._writer.compressedStream ?? this._writer.requestBodyStream, this._nodeTag, this._options.skipOverwriteIfUnchanged);
 
         this._bulkInsertExecuteTask = this._requestExecutor.execute(bulkCommand);
         this._bulkInsertExecuteTask
@@ -955,11 +956,12 @@ export class BulkInsertCommand extends RavenCommand<void> {
     private readonly _stream: Readable;
     private _skipOverwriteIfUnchanged: boolean;
     private readonly _id: number;
-    public useCompression: boolean;
+    private _compressionAlgorithm: HttpCompressionAlgorithm;
 
-    public constructor(id: number, stream: Readable, nodeTag: string, skipOverwriteIfUnchanged: boolean) {
+    public constructor(id: number, compressionAlgorithm: HttpCompressionAlgorithm, stream: Readable, nodeTag: string, skipOverwriteIfUnchanged: boolean) {
         super();
 
+        this._compressionAlgorithm = compressionAlgorithm;
         this._stream = stream;
         this._id = id;
         this._selectedNodeTag = nodeTag;
@@ -973,7 +975,7 @@ export class BulkInsertCommand extends RavenCommand<void> {
             + "&skipOverwriteIfUnchanged=" + (this._skipOverwriteIfUnchanged ? "true" : "false");
 
         const headersBuilder = this._headers().typeAppJson();
-        if (this.useCompression) {
+        if (this._compressionAlgorithm === "Gzip") {
             headersBuilder.with("Content-Encoding", "gzip");
         }
 

src/Documents/Conventions/DocumentConventions.ts

@@ -19,6 +19,7 @@ import { BulkInsertConventions } from "./BulkInsertConventions.js";
 import { InMemoryDocumentSessionOperations } from "../Session/InMemoryDocumentSessionOperations.js";
 import { ShardingConventions } from "./ShardingConventions.js";
 import { plural } from "../../ext/pluralize/pluralize.js";
+import { HttpCompressionAlgorithm } from "../../Http/HttpCompressionAlgorithm.js";
 
 export type IdConvention = (databaseName: string, entity: object) => Promise<string>;
 export type IValueForQueryConverter<T> =
@@ -98,7 +99,9 @@ export class DocumentConventions {
     private _customFetch: any;
     private _dateUtil: DateUtil;
 
-    private _useCompression: boolean;
+    private _useHttpDecompression: boolean | null = null;
+    private _httpCompressionAlgorithm: HttpCompressionAlgorithm = "Gzip";
+
     private _sendApplicationIdentifier: boolean;
 
     private readonly _bulkInsert: BulkInsertConventions;
@@ -156,8 +159,6 @@ export class DocumentConventions {
             documentConventions: this
         });
 
-        this._useCompression = null;
-
         this._dateUtilOpts = {};
         this._dateUtil = new DateUtil(this._dateUtilOpts);
 
@@ -407,10 +408,6 @@ export class DocumentConventions {
         this._maxHttpCacheSize = value;
     }
 
-    public get hasExplicitlySetCompressionUsage() {
-        return this._useCompression !== null;
-    }
-
     public get waitForIndexesAfterSaveChangesTimeout() {
         return this._waitForIndexesAfterSaveChangesTimeout;
     }
@@ -438,17 +435,26 @@ export class DocumentConventions {
         this._waitForReplicationAfterSaveChangesTimeout = value;
     }
 
-    public get useCompression() {
-        if (this._useCompression === null) {
+    /**
+     * Can accept compressed HTTP response content and will use decompression methods
+     */
+    public get useHttpDecompression() {
+        if (this._useHttpDecompression === null) {
             return true;
         }
-
-        return this._useCompression;
+        return this._useHttpDecompression;
     }
 
-    public set useCompression(value) {
+    /**
+     * Can accept compressed HTTP response content and will use decompression methods
+     */
+    public set useHttpDecompression(value: boolean) {
         this._assertNotFrozen();
-        this._useCompression = value;
+        this._useHttpDecompression = value;
+    }
+
+    public get httpCompressionAlgorithm() {
+        return this._httpCompressionAlgorithm;
     }
 
     private _dateUtilOpts: DateUtilOpts;

src/Http/HttpCompressionAlgorithm.ts

@@ -0,0 +1,3 @@
+
+
+export type HttpCompressionAlgorithm = "Gzip";

src/Http/RavenCommand.ts

@@ -168,10 +168,10 @@ export abstract class RavenCommand<TResult> {
         const fetchFn = fetcher ?? fetch; // support for custom fetcher
         const response = await fetchFn(uri, optionsToUse);
 
-        const effectiveStream: Readable =
+        const effectiveStream: Stream =
             response.body
                 ? Readable.fromWeb(response.body)
-                : (response.body ?? new Stream());
+                : new Stream();
 
         effectiveStream
             .pipe(passthrough);

src/Http/RequestExecutor.ts

@@ -365,6 +365,19 @@ export class RequestExecutor implements IDisposable {
         }
     }
 
+    private compressionHeaders(params: HttpRequestParameters) {
+        if (this._conventions.useHttpDecompression) {
+            // do nothing - node.js sends 'accept-encoding: gzip, deflate' by default
+        } else {
+            // disable response compression
+            let { headers} = params;
+            if (!headers) {
+                params.headers = headers = {};
+            }
+            headers["Accept-Encoding"] = "identity";
+        }
+    }
+
     private static async createAgent(options: Agent.Options) {
         try {
             const { Agent: AgentInstance } = await import(importFix("undici"));
@@ -1162,6 +1175,7 @@ export class RequestExecutor implements IDisposable {
         if (this._shouldExecuteOnAll(chosenNode, command)) {
             responseAndStream = await this._executeOnAllToFigureOutTheFastest(chosenNode, command);
         } else {
+            this.compressionHeaders(request);
             responseAndStream = await command.send(await this.getHttpAgent(), request);
         }
 
@@ -1331,6 +1345,7 @@ export class RequestExecutor implements IDisposable {
                         return;
                     }
                     this._setRequestHeaders(null, null, req);
+                    this.compressionHeaders(req);
                     return command.send(agent, req);
                 })
                 .then(commandResult => new IndexAndResponse(taskNumber, commandResult.response, commandResult.bodyStream))
@@ -1918,14 +1933,8 @@ export class RequestExecutor implements IDisposable {
 
 
     private _setDefaultRequestOptions(): void {
-        // add property only if compression enabled - having fixed properly makes custom fetch (like in next.js) fail
-        const compressOpts = !this._conventions.hasExplicitlySetCompressionUsage || this._conventions.useCompression ? {
-            compress: true
-        } : {};
-
         this._defaultRequestOptions = Object.assign(
             DEFAULT_REQUEST_OPTIONS,
-            compressOpts,
             this._customHttpRequestOptions);
     }
 

src/index.ts

@@ -13,6 +13,7 @@ export * from "./Http/IBroadcast.js";
 export * from "./Http/IRaftCommand.js";
 export * from "./Http/NodeSelector.js";
 export * from "./Http/LoadBalanceBehavior.js";
+export * from "./Http/HttpCompressionAlgorithm.js";
 export * from "./Http/RavenCommand.js";
 export * from "./Http/ReadBalanceBehavior.js";
 export * from "./Http/RequestExecutor.js";

test/Executor/CompressionTests.ts

@@ -6,6 +6,7 @@ import DocumentStore, {
     IDocumentStore,
     GetDatabaseNamesCommand,
 } from "../../src/index.js";
+import { assertThat } from "../Utils/AssertExtensions.js";
 
 describe("Compression", function () {
 
@@ -25,16 +26,35 @@ describe("Compression", function () {
 
         await exec.execute(cmd);
         const reqParams = createReqSpy.lastCall.returnValue;
-        //TODO: assert.ok(reqParams.compress);
+        assertThat(reqParams.headers["Accept-Encoding"])
+            .isNull();
     });
 
-    it("is turned off on demand", async () => {
+    it("can be turned on on demand", async () => {
         const store2 = new DocumentStore(store.urls, store.database);
         try {
-            store2.conventions.useCompression = false;
-            assert.ok(store2.conventions.hasExplicitlySetCompressionUsage);
-            assert.ok(!store2.conventions.useCompression);
+            store2.conventions.useHttpDecompression = true;
+            store2.initialize();
+
+            const exec = store2.getRequestExecutor();
+            const cmd = new GetDatabaseNamesCommand(0, 5);
+            const createReqSpy = exec["_createRequest"] = sinon.spy(exec["_createRequest"]);
+
+            await exec.execute(cmd);
+            const reqParams = createReqSpy.lastCall.returnValue;
+            assertThat(reqParams.headers["Accept-Encoding"])
+                .isNull();
+        } finally {
+            if (store2) {
+                store2.dispose();
+            }
+        }
+    });
 
+    it("can be turned off on demand", async () => {
+        const store2 = new DocumentStore(store.urls, store.database);
+        try {
+            store2.conventions.useHttpDecompression = false;
             store2.initialize();
 
             const exec = store2.getRequestExecutor();
@@ -43,7 +63,8 @@ describe("Compression", function () {
 
             await exec.execute(cmd);
             const reqParams = createReqSpy.lastCall.returnValue;
-            //TODO: assert.ok(!reqParams.compress);
+            assertThat(reqParams.headers["Accept-Encoding"])
+                .isEqualTo("identity");
         } finally {
             if (store2) {
                 store2.dispose();

tsconfig.json

@@ -16,7 +16,6 @@
     "lib": [ "ES2022" ],
     "typeRoots": [
       "node_modules/@types",
-      "node_modules/moment",
       "src/Types"
     ],
     "types": [