Merge pull request #360 from ml054/v5.2

RDBC-654 Investigate cloudflare and node.js client interoperability

Author: ml054

.github/workflows/RavenClient.yml

@@ -24,7 +24,7 @@ jobs:
 
     strategy:
       matrix:
-        node-version: [12.x, 14.x, 16.x, 18.x]
+        node-version: [14.x, 16.x, 18.x]
         serverVersion: ["5.2", "5.4"]
       fail-fast: false
 

package.json

@@ -41,7 +41,7 @@
     "Hibernating Rhinos"
   ],
   "engines": {
-    "node": ">=6.0.0"
+    "node": ">=14.0.0"
   },
   "keywords": [
     "ravendb",

src/Documents/Conventions/DocumentConventions.ts

@@ -93,6 +93,7 @@ export class DocumentConventions {
     private _remoteEntityFieldNameConvention: CasingConvention;
 
     private _objectMapper: TypesAwareObjectMapper;
+    private _customFetch: any;
     private _dateUtil: DateUtil;
     private _syncJsonParseLimit: number;
 
@@ -109,7 +110,7 @@ export class DocumentConventions {
         this._readBalanceBehavior = "None";
         this._identityPartsSeparator = "/";
         this._identityProperty = CONSTANTS.Documents.Metadata.ID_PROPERTY;
-        
+
         this._findJsType = (id: string, doc: object) => {
             const metadata = doc[CONSTANTS.Documents.Metadata.KEY];
             if (metadata) {
@@ -249,9 +250,28 @@ export class DocumentConventions {
         this._objectMapper = value;
     }
 
+    public get customFetch(): any {
+        return this._customFetch;
+    }
+
+    /**
+     * Allows to override default fetch method
+     *
+     * This method is useful to enable RavenDB node.js client
+     * on CloudFlare Workers
+     *
+     * You should pass object bound to worker with type: mtls_certificate
+     *
+     * @param customFetch
+     */
+    public set customFetch(customFetch: any) {
+        this._assertNotFrozen();
+        this._customFetch = customFetch;
+    }
+
     /**
      * Sets json length limit for sync parsing. Beyond that size
-     * we fallback to async parsing
+     * we fall back to async parsing
      */
     public get syncJsonParseLimit(): number {
         return this._syncJsonParseLimit;
@@ -435,7 +455,7 @@ export class DocumentConventions {
     public set storeDatesWithTimezoneInfo(value) {
         this._assertNotFrozen();
         this._dateUtilOpts.withTimezone = true;
-    } 
+    }
 
     /**
      * If set to 'true' then it will throw an exception when any query is performed (in session)

src/Http/RavenCommand.ts

@@ -13,7 +13,9 @@ import { JsonSerializer } from "../Mapping/Json/Serializer";
 import { RavenCommandResponsePipeline } from "./RavenCommandResponsePipeline";
 import { DocumentConventions } from "../Documents/Conventions/DocumentConventions";
 import * as http from "http";
-import { ObjectTypeDescriptor, ServerResponse } from "../Types";
+import { ObjectTypeDescriptor } from "../Types";
+import { ReadableWebToNodeStream } from "../Utility/ReadableWebToNodeStream";
+import { LengthUnawareFormData } from "../Utility/LengthUnawareFormData";
 
 const log = getLogger({ module: "RavenCommand" });
 
@@ -120,19 +122,27 @@ export abstract class RavenCommand<TResult> {
 
     public async send(agent: http.Agent,
         requestOptions: HttpRequestParameters): Promise<{ response: HttpResponse, bodyStream: stream.Readable }> {
-        const { body, uri, ...restOptions } = requestOptions;
+
+        const { body, uri, fetcher, ...restOptions } = requestOptions;
+
         log.info(`Send command ${this.constructor.name} to ${uri}${body ? " with body " + body : ""}.`);
 
         if (requestOptions.agent) { // support for fiddler
             agent = requestOptions.agent as http.Agent;
         }
 
-        const optionsToUse = { body, ...restOptions, agent } as RequestInit;
+        const bodyToUse = fetcher ? RavenCommand.maybeWrapBody(body) : body;
+
+        const optionsToUse = { body: bodyToUse, ...restOptions, agent } as RequestInit;
 
         const passthrough = new stream.PassThrough();
-        const response = await fetch(uri, optionsToUse);
         passthrough.pause();
-        response.body
+
+        const fetchFn = fetcher ?? fetch; // support for custom fetcher
+        const response = await fetchFn(uri, optionsToUse);
+
+        const effectiveStream: stream.Readable = fetcher ? new ReadableWebToNodeStream(response.body) : response.body;
+        effectiveStream
             .pipe(passthrough);
 
         return {
@@ -141,6 +151,17 @@ export abstract class RavenCommand<TResult> {
         };
     }
 
+    private static maybeWrapBody(body: any) {
+        if (body instanceof LengthUnawareFormData) {
+            throw new Error("Requests using FormData as payload are not yet supported!");
+        }
+        if (body instanceof stream.Readable) {
+            throw new Error("Requests using stream.Readable as payload are not yet supported!");
+        }
+
+        return body;
+    }
+
     public setResponseRaw(response: HttpResponse, body: string): void {
         throwError("NotSupportedException",
             "When _responseType is set to RAW then please override this method to handle the response.");

src/Http/RequestExecutor.ts

@@ -33,13 +33,12 @@ import { validateUri } from "../Utility/UriUtil";
 import * as StreamUtil from "../Utility/StreamUtil";
 import { closeHttpResponse } from "../Utility/HttpUtil";
 import { PromiseStatusTracker } from "../Utility/PromiseUtil";
-import * as http from "http";
-import * as https from "https";
+import type * as http from "http";
+import type * as https from "https";
 import { IBroadcast } from "./IBroadcast";
 import { StringUtil } from "../Utility/StringUtil";
 import { IRaftCommand } from "./IRaftCommand";
 import AbortController from "abort-controller";
-import { URL } from "url";
 import { EventEmitter } from "events";
 import {
     BeforeRequestEventArgs,
@@ -142,7 +141,11 @@ export class NodeStatus implements IDisposable {
 export class RequestExecutor implements IDisposable {
     private _emitter = new EventEmitter();
 
-    private static readonly GLOBAL_APPLICATION_IDENTIFIER = uuidv4().toString();
+    /*
+      we don't initialize this here due to issue with cloudflare
+      see: https://github.com/cloudflare/miniflare/issues/292
+     */
+    private static GLOBAL_APPLICATION_IDENTIFIER: string = null;
 
     private static readonly INITIAL_TOPOLOGY_ETAG = -2;
 
@@ -183,9 +186,11 @@ export class RequestExecutor implements IDisposable {
 
     private _httpAgent: http.Agent;
 
-    private static readonly KEEP_ALIVE_HTTP_AGENT = new http.Agent({
-        keepAlive: true
-    });
+    /*
+      we don't initialize this here due to issue with cloudflare
+      see: https://github.com/cloudflare/miniflare/issues/292
+    */
+    private static KEEP_ALIVE_HTTP_AGENT: http.Agent = null;
 
     private static readonly HTTPS_AGENT_CACHE = new Map<string, https.Agent>();
 
@@ -330,6 +335,10 @@ export class RequestExecutor implements IDisposable {
     }
 
     public getHttpAgent(): http.Agent {
+        if (this.conventions.customFetch) {
+            return null;
+        }
+
         if (this._httpAgent) {
             return this._httpAgent;
         }
@@ -344,6 +353,9 @@ export class RequestExecutor implements IDisposable {
             if (RequestExecutor.HTTPS_AGENT_CACHE.has(cacheKey)) {
                 return RequestExecutor.HTTPS_AGENT_CACHE.get(cacheKey);
             } else {
+                // eslint-disable-next-line @typescript-eslint/no-var-requires
+                const https = require("https");
+
                 const agent = new https.Agent({
                     keepAlive: true,
                     ...agentOptions
@@ -353,10 +365,22 @@ export class RequestExecutor implements IDisposable {
                 return agent;
             }
         } else {
+            RequestExecutor.assertKeepAliveAgent();
             return RequestExecutor.KEEP_ALIVE_HTTP_AGENT;
         }
     }
 
+    private static assertKeepAliveAgent() {
+        if (!RequestExecutor.KEEP_ALIVE_HTTP_AGENT) {
+            // eslint-disable-next-line @typescript-eslint/no-var-requires
+            const http = require("http");
+
+            RequestExecutor.KEEP_ALIVE_HTTP_AGENT = new http.Agent({
+                keepAlive: true
+            });
+        }
+    }
+
     public getTopologyNodes(): ServerNode[] {
         const topology = this.getTopology();
         return topology
@@ -399,14 +423,25 @@ export class RequestExecutor implements IDisposable {
         opts?: IRequestExecutorOptions): RequestExecutor {
         const { authOptions, documentConventions } = opts || {} as IRequestExecutorOptions;
         const executor = new RequestExecutor(database, authOptions, documentConventions);
-        executor._firstTopologyUpdatePromise = executor._firstTopologyUpdate(initialUrls, this.GLOBAL_APPLICATION_IDENTIFIER);
+
+        executor._firstTopologyUpdatePromise = executor._firstTopologyUpdate(initialUrls, RequestExecutor.getGlobalApplicationIdentifier());
 
         // this is just to get rid of unhandled rejection, we're handling it later on
         executor._firstTopologyUpdatePromise.catch(TypeUtil.NOOP);
 
         return executor;
     }
 
+    private static getGlobalApplicationIdentifier() {
+        // due to cloudflare constraints we can't init GLOBAL_APPLICATION_IDENTIFIER in static
+
+        if (!this.GLOBAL_APPLICATION_IDENTIFIER) {
+            this.GLOBAL_APPLICATION_IDENTIFIER = uuidv4();
+        }
+
+        return this.GLOBAL_APPLICATION_IDENTIFIER;
+    }
+
     public static createForSingleNodeWithConfigurationUpdates(
         url: string, database: string, opts: IRequestExecutorOptions): RequestExecutor {
         const executor =
@@ -1284,6 +1319,10 @@ export class RequestExecutor implements IDisposable {
             return null;
         }
 
+        if (this.conventions.customFetch) {
+            request.fetcher = this.conventions.customFetch;
+        }
+
         const req = Object.assign(request, this._defaultRequestOptions);
         urlRef(req.uri);
         req.headers = req.headers || {};

src/Primitives/Http.ts

@@ -1,6 +1,11 @@
 import { Request, RequestInit, Response } from "node-fetch";
 
-export type HttpRequestParameters = RequestInit & { uri: string };
-export type HttpRequestParametersWithoutUri = RequestInit;
+export type HttpRequestParameters = RequestInit & {
+  uri: string;
+  fetcher?: any;
+};
+export type HttpRequestParametersWithoutUri = RequestInit & {
+  fetcher?: any;
+};
 export type HttpResponse = Response;
 export type HttpRequest = Request;

src/Utility/ReadableWebToNodeStream.ts

@@ -0,0 +1,79 @@
+// Taken from: https://github.com/Borewit/readable-web-to-node-stream/blob/master/lib/index.ts
+
+import { Readable } from 'readable-stream';
+
+/**
+ * Converts a Web-API stream into Node stream.Readable class
+ * Node stream readable: https://nodejs.org/api/stream.html#stream_readable_streams
+ * Web API readable-stream: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream
+ * Node readable stream: https://nodejs.org/api/stream.html#stream_readable_streams
+ */
+export class ReadableWebToNodeStream extends Readable {
+
+    public bytesRead: number = 0;
+    public released = false;
+
+    /**
+     * Default web API stream reader
+     * https://developer.mozilla.org/en-US/docs/Web/API/ReadableStreamDefaultReader
+     */
+    private reader: any;
+    private pendingRead: Promise<any> | undefined;
+
+    /**
+     *
+     * @param stream ReadableStream: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream
+     */
+    constructor(stream: any) {
+        super();
+        this.reader = stream.getReader();
+    }
+
+    /**
+     * Implementation of readable._read(size).
+     * When readable._read() is called, if data is available from the resource,
+     * the implementation should begin pushing that data into the read queue
+     * https://nodejs.org/api/stream.html#stream_readable_read_size_1
+     */
+    public async _read() {
+        // Should start pushing data into the queue
+        // Read data from the underlying Web-API-readable-stream
+        if (this.released) {
+            this.push(null); // Signal EOF
+            return;
+        }
+        this.pendingRead = this.reader.read();
+        const data = await this.pendingRead;
+        // clear the promise before pushing new data to the queue and allow sequential calls to _read()
+        delete this.pendingRead;
+        if (data.done || this.released) {
+            this.push(null); // Signal EOF
+        } else {
+            this.bytesRead += data.value.length;
+            this.push(data.value); // Push new data to the queue
+        }
+    }
+
+    /**
+     * If there is no unresolved read call to Web-API Readable Stream immediately returns;
+     * otherwise will wait until the read is resolved.
+     */
+    public async waitForReadToComplete() {
+        if (this.pendingRead) {
+            await this.pendingRead;
+        }
+    }
+
+    /**
+     * Close wrapper
+     */
+    public async close(): Promise<void> {
+        await this.syncAndRelease();
+    }
+
+    private async syncAndRelease() {
+        this.released = true;
+        await this.waitForReadToComplete();
+        await this.reader.releaseLock();
+    }
+}