RDBC-622 parse patch in sync way

Author: ml054

src/Documents/Commands/MultiGet/MultiGetCommand.ts

@@ -14,6 +14,10 @@ import { IDisposable } from "../../../Types/Contracts";
 import { RequestExecutor } from "../../../Http/RequestExecutor";
 import { AggressiveCacheOptions } from "../../../Http/AggressiveCacheOptions";
 import { HEADERS } from "../../../Constants";
+import { ServerCasing, ServerResponse } from "../../../Types";
+import { ConditionalGetResult } from "../ConditionalGetDocumentsCommand";
+import { ObjectUtil } from "../../../Utility/ObjectUtil";
+import { camelCase } from "change-case";
 
 export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDisposable {
     private readonly _requestExecutor: RequestExecutor;
@@ -144,27 +148,11 @@ export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDis
             this._throwInvalidResponse();
         }
         try {
-            const result = await this._pipeline<GetResponse[]>()
-                .parseJsonAsync()
-                .jsonKeysTransform({
-                    getCurrentTransform(key, stack) {
-                        if (stack.length === 1
-                            || stack.length === 2
-                            || stack.length === 3) {
-                            // results.0.result
-                            return "camel";
-                        }
-
-                        return null;
-                    }
-                })
+            const result = await this._pipeline<ServerCasing<ServerResponse<{ Results: GetResponse[] }>>>()
+                .parseJsonSync()
                 .process(bodyStream);
 
-            const responses = result["results"].reduce((result: GetResponse[], next) => {
-                // TODO try to get it directly from parser
-                next.result = TypeUtil.isNullOrUndefined(next.result) ? next.result : JSON.stringify(next.result);
-                return [...result, next];
-            }, []);
+            const responses = result.Results.map(item => MultiGetCommand._mapToLocalObject(item));
 
             this.result = [];
 
@@ -243,6 +231,17 @@ export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDis
             }
         }
     }
+
+    private static _mapToLocalObject(json: ServerCasing<ServerResponse<GetResponse>>): any {
+        // convert from Pascal to camel on top level only
+        const item: any = {};
+        for (const [key, value] of Object.entries(json)) {
+            item[camelCase(key)] = value;
+        }
+
+        item.result = item.result ? JSON.stringify(item.result) : null;
+        return item;
+    }
 }
 
 class Cached implements IDisposable {

src/Documents/Operations/PatchOperation.ts

@@ -11,6 +11,7 @@ import { ServerNode } from "../../Http/ServerNode";
 import { PatchResult } from "./PatchResult";
 import * as stream from "readable-stream";
 import { ObjectUtil } from "../../Utility/ObjectUtil";
+import { ServerCasing, ServerResponse } from "../../Types";
 
 export interface Payload {
     patch: PatchRequest;
@@ -175,11 +176,25 @@ export class PatchCommand extends RavenCommand<PatchResult> {
         }
 
         let body;
-        this.result = await this._pipeline<PatchResult>()
+        const results= await this._pipeline<ServerCasing<ServerResponse<PatchResult>>>()
             .collectBody(_ => body = _)
-            .parseJsonAsync()
-            .jsonKeysTransform("Patch", this._conventions)
+            .parseJsonSync()
             .process(bodyStream);
+
+        this.result = PatchCommand._mapToLocalObject(results, this._conventions);
+
         return body;
     }
+
+    private static _mapToLocalObject(json: ServerCasing<ServerResponse<PatchResult>>, conventions: DocumentConventions): PatchResult {
+        return {
+            changeVector: json.ChangeVector,
+            collection: json.Collection,
+            debug: json.Debug,
+            lastModified: conventions.dateUtil.parse(json.LastModified),
+            status: json.Status,
+            modifiedDocument: ObjectUtil.transformDocumentKeys(json.ModifiedDocument, conventions),
+            originalDocument: ObjectUtil.transformDocumentKeys(json.OriginalDocument, conventions)
+        }
+    }
 }

src/Http/RavenCommandResponsePipeline.ts

@@ -17,14 +17,6 @@ import {
     lastChunk
 } from "../Mapping/Json/Streams/CollectResultStream";
 import { throwError, getError } from "../Exceptions";
-import {
-    TransformJsonKeysStreamOptions,
-    TransformKeysJsonStream
-} from "../Mapping/Json/Streams/TransformKeysJsonStream";
-import {
-    getTransformJsonKeysProfile,
-    TransformJsonKeysProfile
-} from "../Mapping/Json/Streams/TransformJsonKeysProfiles";
 import { TypeUtil } from "../Utility/TypeUtil";
 import * as Asm from "stream-json/Assembler";
 import { DocumentConventions } from "../Documents/Conventions/DocumentConventions";
@@ -44,7 +36,6 @@ export interface RavenCommandResponsePipelineOptions<TResult> {
     streamKeyCaseTransform?: ObjectKeyCaseTransformStreamOptions;
     collectResult: CollectResultStreamOptions<TResult>;
     transform?: stream.Stream;
-    transformKeys?: TransformJsonKeysStreamOptions;
 }
 
 export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {
@@ -75,20 +66,16 @@ export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {
      * @param type Type of object to extract from objects stream - use Raw to skip extraction.
      * @param options
      */
-    public parseJsonlAsync(valueExtractor: (obj: any) => any, conventions?: DocumentConventions, options: { transforms?: stream.Transform[] } = {}) {
+    public parseJsonlAsync(valueExtractor: (obj: any) => any, options: { transforms?: stream.Transform[] } = {}) {
         const transforms = options?.transforms ?? [];
         const extractItemTransform = new stream.Transform({
             objectMode: true,
             transform(chunk, encoding, callback) {
-                let value = valueExtractor(chunk["value"]);
+                const value = valueExtractor(chunk["value"]);
                 if (!value) {
                     return callback();
                 }
 
-                if (conventions) {
-                    value = ObjectUtil.transformDocumentKeys(value, conventions);
-                }
-
                 callback(null, {...chunk, value});
             }
         });
@@ -107,32 +94,6 @@ export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {
         return this;
     }
 
-    public jsonKeysTransform(): this;
-    public jsonKeysTransform(profile: TransformJsonKeysProfile, conventions: DocumentConventions): this;
-    public jsonKeysTransform(profile: TransformJsonKeysProfile): this;
-    public jsonKeysTransform(opts: TransformJsonKeysStreamOptions): this;
-    public jsonKeysTransform(
-        optsOrProfile?: TransformJsonKeysStreamOptions | TransformJsonKeysProfile,
-        conventions?: DocumentConventions): this {
-
-        if (!this._opts.jsonAsync) {
-            throwError("InvalidOperationException",
-                "Cannot use transformKeys without doing parseJsonAsync() first.");
-        }
-
-        if (!optsOrProfile) {
-            throwError("InvalidArgumentException", "Must provide transform opts or profile name.");
-        }
-
-        if (TypeUtil.isString(optsOrProfile)) {
-            this._opts.transformKeys = getTransformJsonKeysProfile(optsOrProfile, conventions);
-        } else {
-            this._opts.transformKeys = optsOrProfile;
-        }
-
-        return this;
-    }
-
     public objectKeysTransform(defaultTransform: CasingConvention, profile?: ObjectKeyCaseTransformProfile): this;
     public objectKeysTransform(opts: ObjectKeyCaseTransformStreamOptions): this;
     public objectKeysTransform(
@@ -248,10 +209,6 @@ export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {
             streams.push(new ObjectKeyCaseTransformStream(keyCaseOpts));
         }
 
-        if (opts.transformKeys) {
-            streams.push(new TransformKeysJsonStream(opts.transformKeys) as any);
-        }
-
         return streams;
     }
 

src/Mapping/Json/Streams/Pipelines.ts

@@ -3,23 +3,40 @@ import { RavenCommandResponsePipeline } from "../../../Http/RavenCommandResponse
 import { DocumentConventions } from "../../../Documents/Conventions/DocumentConventions";
 import { stringer as jsonlStringer } from "stream-json/jsonl/Stringer";
 import { stringer } from "stream-json/Stringer";
-import { TransformKeysJsonStream } from "./TransformKeysJsonStream";
-import { getTransformJsonKeysProfile } from "./TransformJsonKeysProfiles";
 import { pick } from "stream-json/filters/Pick";
 import { streamArray } from "stream-json/streamers/StreamArray";
+import { ObjectUtil } from "../../../Utility/ObjectUtil";
 
 export function getDocumentResultsAsObjects(
     conventions: DocumentConventions,
     queryStream: boolean
 ): RavenCommandResponsePipeline<object[]> {
     const pipeline = RavenCommandResponsePipeline.create<object[]>();
 
+    const keysTransform = new stream.Transform({
+        objectMode: true,
+        transform(chunk, encoding, callback) {
+            let value = chunk["value"];
+            if (!value) {
+                return callback();
+            }
+
+            if (conventions) {
+                value = ObjectUtil.transformDocumentKeys(value, conventions);
+            }
+
+            callback(null, {...chunk, value});
+        }
+    });
+
     return conventions.useJsonlStreaming
-        ? pipeline.parseJsonlAsync(queryStream ? x => x["Item"] : x => x, conventions)
+        ? pipeline.parseJsonlAsync(queryStream ? x => x["Item"] : x => x, {
+            transforms: [keysTransform]
+        })
         : pipeline.parseJsonAsync([
-            new TransformKeysJsonStream(getTransformJsonKeysProfile("DocumentLoad", conventions)),
-            pick({ filter: "results" }),
-            streamArray()
+            pick({ filter: "Results" }),
+            streamArray(),
+            keysTransform
         ]);
 }
 
@@ -29,14 +46,13 @@ export function getDocumentStreamResultsIntoStreamPipeline(
     const pipeline = RavenCommandResponsePipeline.create<object[]>();
 
     return conventions.useJsonlStreaming
-        ? pipeline.parseJsonlAsync(x => x["Item"], conventions, {
+        ? pipeline.parseJsonlAsync(x => x["Item"], {
             transforms: [
                 jsonlStringer({ replacer: (key, value) => key === '' ? value.value : value }),
             ]
         })
         : pipeline
             .parseJsonAsync([
-                new TransformKeysJsonStream(getTransformJsonKeysProfile("DocumentLoad", conventions)),
                 stringer({ useValues: true })
             ]);
 }