RDBC-622 parse cmp exchange in sync way

ml054 · ml054 · commit 2898bfe005de · 2023-08-08T13:52:23.000+02:00
diff --git a/src/Documents/Commands/FacetQueryCommand.ts b/src/Documents/Commands/FacetQueryCommand.ts
@@ -48,12 +48,19 @@ export class FacetQueryCommand extends QueryCommand {
             defaultTransform: "camel"
         }) as any;
 
+        const mappedIncludes: Record<string, any> = {};
+        if (json.Includes) {
+            for (const [key, value] of Object.entries(json.Includes)) {
+                mappedIncludes[key] = ObjectUtil.transformDocumentKeys(value, conventions);
+            }
+        }
+
         return {
             ...restMapped,
             indexTimestamp: conventions.dateUtil.parse(IndexTimestamp),
             lastQueryTime: conventions.dateUtil.parse(LastQueryTime),
             results: Results.map(x => ObjectUtil.transformObjectKeys(x, { defaultTransform: "camel" })),
-            includes: {}
+            includes: mappedIncludes
         };
     }
 }
diff --git a/src/Documents/Conventions/DocumentConventions.ts b/src/Documents/Conventions/DocumentConventions.ts
@@ -102,7 +102,7 @@ export class DocumentConventions {
 
     private readonly _bulkInsert: BulkInsertConventions;
 
-    private _useJsonlStreaming = false;
+    private _useJsonlStreaming = true;
 
     public get bulkInsert() {
         return this._bulkInsert;
diff --git a/src/Documents/Operations/CompareExchange/CompareExchangeValueResultParser.ts b/src/Documents/Operations/CompareExchange/CompareExchangeValueResultParser.ts
@@ -10,7 +10,7 @@ import { CompareExchangeResultClass, EntityConstructor } from "../../../Types";
 export interface CompareExchangeResultItem {
     index: number;
     key: string;
-    value: { object: object };
+    value: { object: object, "@metadata"?: any };
 }
 
 export interface GetCompareExchangeValuesResponse {
@@ -91,7 +91,7 @@ export class CompareExchangeValueResultParser {
         }
 
         let rawValue = raw.object;
-        if (TypeUtil.isPrimitiveType(clazz) || !clazz) {
+        if (clazz && TypeUtil.isPrimitiveType(clazz) || TypeUtil.isPrimitive(rawValue)) {
             return new CompareExchangeValue(key, index, rawValue, metadata);
         } else {
             if (!rawValue) {
diff --git a/src/Documents/Operations/CompareExchange/GetCompareExchangeValueOperation.ts b/src/Documents/Operations/CompareExchange/GetCompareExchangeValueOperation.ts
@@ -7,9 +7,10 @@ import { DocumentConventions } from "../../Conventions/DocumentConventions";
 import { IDocumentStore } from "../../IDocumentStore";
 import { throwError } from "../../../Exceptions";
 import { ServerNode } from "../../../Http/ServerNode";
-import { CompareExchangeResultClass } from "../../../Types";
+import { CompareExchangeResultClass, ServerCasing, ServerResponse } from "../../../Types";
 import { CompareExchangeValueResultParser, GetCompareExchangeValuesResponse } from "./CompareExchangeValueResultParser";
 import * as stream from "readable-stream";
+import { GetCompareExchangeValuesCommand } from "./GetCompareExchangeValuesOperation";
 
 export class GetCompareExchangeValueOperation<T> implements IOperation<CompareExchangeValue<T>> {
 
@@ -60,7 +61,7 @@ export class GetCompareExchangeValueCommand<T> extends RavenCommand<CompareExcha
 
     public createRequest(node: ServerNode): HttpRequestParameters {
         const uri = node.url + "/databases/" + node.database + "/cmpxchg?key=" + encodeURIComponent(this._key);
-        return { uri };
+        return {uri};
     }
 
     public async setResponseAsync(bodyStream: stream.Stream, fromCache: boolean): Promise<string> {
@@ -69,14 +70,14 @@ export class GetCompareExchangeValueCommand<T> extends RavenCommand<CompareExcha
         }
 
         let body: string = null;
-        const results = await this._pipeline<GetCompareExchangeValuesResponse>()
+        const results = await this._pipeline<ServerCasing<ServerResponse<GetCompareExchangeValuesResponse>>>()
             .collectBody(x => body = x)
-            .parseJsonAsync()
-            .jsonKeysTransform("GetCompareExchangeValue", this._conventions)
+            .parseJsonSync()
             .process(bodyStream);
 
-        this.result = CompareExchangeValueResultParser.getValue(
-                results as GetCompareExchangeValuesResponse, this._materializeMetadata, this._conventions, this._clazz);
+        const localObject = GetCompareExchangeValuesCommand.mapToLocalObject(results);
+
+        this.result = CompareExchangeValueResultParser.getValue(localObject, this._materializeMetadata, this._conventions, this._clazz);
         return body;
     }
 }
diff --git a/src/Documents/Operations/CompareExchange/GetCompareExchangeValuesOperation.ts b/src/Documents/Operations/CompareExchange/GetCompareExchangeValuesOperation.ts
@@ -2,7 +2,7 @@ import { HttpRequestParameters } from "../../../Primitives/Http";
 import { IOperation, OperationResultType } from "../OperationAbstractions";
 import { CompareExchangeValue } from "./CompareExchangeValue";
 import { throwError } from "../../../Exceptions";
-import { CompareExchangeResultClass } from "../../../Types";
+import { CompareExchangeResultClass, ServerCasing, ServerResponse } from "../../../Types";
 import { IDocumentStore } from "../../IDocumentStore";
 import { DocumentConventions } from "../../Conventions/DocumentConventions";
 import { HttpCache } from "../../../Http/HttpCache";
@@ -12,6 +12,7 @@ import { CompareExchangeValueResultParser, GetCompareExchangeValuesResponse } fr
 import * as stream from "readable-stream";
 import { StringBuilder } from "../../../Utility/StringBuilder";
 import { TypeUtil } from "../../../Utility/TypeUtil";
+import { QueryResult } from "../../Queries/QueryResult";
 
 export interface GetCompareExchangeValuesParameters<T> {
     keys?: string[];
@@ -142,15 +143,41 @@ export class GetCompareExchangeValuesCommand<T> extends RavenCommand<{ [key: str
             return body;
         }
 
-        const results = await this._pipeline<GetCompareExchangeValuesResponse>()
+        const results = await this._pipeline<ServerCasing<ServerResponse<GetCompareExchangeValuesResponse>>>()
             .collectBody(b => body = b)
-            .parseJsonAsync()
-            .jsonKeysTransform("GetCompareExchangeValue", this._conventions)
+            .parseJsonSync()
             .process(bodyStream);
 
+        const localObject = GetCompareExchangeValuesCommand.mapToLocalObject(results);
+
         this.result = CompareExchangeValueResultParser.getValues<T>(
-            results, this._materializeMetadata, this._conventions, this._operation.clazz);
+            localObject, this._materializeMetadata, this._conventions, this._operation.clazz);
 
         return body;
     }
+
+    public static mapToLocalObject(json: ServerCasing<ServerResponse<GetCompareExchangeValuesResponse>>): GetCompareExchangeValuesResponse {
+        return {
+            results: json.Results.map(item => {
+                if (!item.Value) {
+                    return {
+                        index: item.Index,
+                        key: item.Key,
+                        value: null
+                    }
+                }
+
+                const { Object, ...rest} = item.Value;
+
+                return {
+                    value: {
+                        ...rest,
+                        object: Object
+                    },
+                    index: item.Index,
+                    key: item.Key
+                }
+            })
+        }
+    }
 }
diff --git a/src/Documents/Session/Operations/Lazy/LazyGetCompareExchangeValueOperation.ts b/src/Documents/Session/Operations/Lazy/LazyGetCompareExchangeValueOperation.ts
@@ -10,9 +10,10 @@ import {
     CompareExchangeValueResultParser,
     GetCompareExchangeValuesResponse
 } from "../../../Operations/CompareExchange/CompareExchangeValueResultParser";
-import { CompareExchangeResultClass } from "../../../../Types";
+import { CompareExchangeResultClass, ServerCasing, ServerResponse } from "../../../../Types";
 import { DocumentConventions } from "../../../Conventions/DocumentConventions";
 import { QueryResult } from "../../../Queries/QueryResult";
+import { GetCompareExchangeValuesCommand } from "../../../Operations/CompareExchange/GetCompareExchangeValuesOperation";
 
 export class LazyGetCompareExchangeValueOperation<T> implements ILazyOperation {
     private readonly _clusterSession: ClusterTransactionOperationsBase;
@@ -76,12 +77,11 @@ export class LazyGetCompareExchangeValueOperation<T> implements ILazyOperation {
         }
 
         if (response.result) {
-            const results = await RavenCommandResponsePipeline.create<GetCompareExchangeValuesResponse>()
-                .parseJsonAsync()
-                .jsonKeysTransform("GetCompareExchangeValue", this._conventions)
-                .process(stringToReadable(response.result));
+            const results = JSON.parse(response.result) as ServerCasing<ServerResponse<GetCompareExchangeValuesResponse>>;
 
-            const value = CompareExchangeValueResultParser.getValue(results, false, this._conventions, null);
+            const localObject = GetCompareExchangeValuesCommand.mapToLocalObject(results);
+
+            const value = CompareExchangeValueResultParser.getValue(localObject, false, this._conventions, null);
 
             if (this._clusterSession.session.noTracking) {
                 if (!value) {
diff --git a/src/Documents/Session/Operations/Lazy/LazyGetCompareExchangeValuesOperation.ts b/src/Documents/Session/Operations/Lazy/LazyGetCompareExchangeValuesOperation.ts
@@ -1,6 +1,6 @@
 import { ILazyOperation } from "./ILazyOperation";
 import { ClusterTransactionOperationsBase } from "../../ClusterTransactionOperationsBase";
-import { CompareExchangeResultClass } from "../../../../Types";
+import { CompareExchangeResultClass, ServerCasing, ServerResponse } from "../../../../Types";
 import { throwError } from "../../../../Exceptions";
 import { TypeUtil } from "../../../../Utility/TypeUtil";
 import { GetRequest } from "../../../Commands/MultiGet/GetRequest";
@@ -16,6 +16,7 @@ import { DocumentConventions } from "../../../Conventions/DocumentConventions";
 import { QueryResult } from "../../../Queries/QueryResult";
 import { CompareExchangeValue } from "../../../Operations/CompareExchange/CompareExchangeValue";
 import { StringBuilder } from "../../../../Utility/StringBuilder";
+import { GetCompareExchangeValuesCommand } from "../../../Operations/CompareExchange/GetCompareExchangeValuesOperation";
 
 export class LazyGetCompareExchangeValuesOperation<T> implements ILazyOperation {
     private readonly _clusterSession: ClusterTransactionOperationsBase;
@@ -131,14 +132,13 @@ export class LazyGetCompareExchangeValuesOperation<T> implements ILazyOperation
         }
 
         if (response.result) {
-            const results = await RavenCommandResponsePipeline.create<GetCompareExchangeValuesResponse>()
-                .parseJsonAsync()
-                .jsonKeysTransform("GetCompareExchangeValue", this._conventions)
-                .process(stringToReadable(response.result));
+            const results = JSON.parse(response.result) as ServerCasing<ServerResponse<GetCompareExchangeValuesResponse>>;
+
+            const localObject = GetCompareExchangeValuesCommand.mapToLocalObject(results);
 
             if (this._clusterSession.session.noTracking) {
                 const result: { [key: string]: CompareExchangeValue<T> } = {};
-                for (const kvp of Object.entries(CompareExchangeValueResultParser.getValues(results, false, this._conventions))) {
+                for (const kvp of Object.entries(CompareExchangeValueResultParser.getValues(localObject, false, this._conventions))) {
                     if (!kvp[1].value) {
                         result[kvp[0]] = this._clusterSession.registerMissingCompareExchangeValue(kvp[0]).getValue(this._clazz, this._conventions);
                         continue;
@@ -151,7 +151,7 @@ export class LazyGetCompareExchangeValuesOperation<T> implements ILazyOperation
                 return;
             }
 
-            for (const kvp of Object.entries(CompareExchangeValueResultParser.getValues(results, false, this._conventions))) {
+            for (const kvp of Object.entries(CompareExchangeValueResultParser.getValues(localObject, false, this._conventions))) {
                 if (!kvp[1]) {
                     continue;
                 }
diff --git a/src/Http/RavenCommandResponsePipeline.ts b/src/Http/RavenCommandResponsePipeline.ts
@@ -93,7 +93,7 @@ export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {
             }
         });
 
-        transforms.push(extractItemTransform);
+        transforms.unshift(extractItemTransform);
 
         this._opts.jsonlAsync = {
             transforms
diff --git a/src/Mapping/Json/Streams/TransformJsonKeysProfiles.ts b/src/Mapping/Json/Streams/TransformJsonKeysProfiles.ts
@@ -6,7 +6,6 @@ export type TransformJsonKeysProfile =
     "DocumentLoad"
     | "Patch"
     | "CompareExchangeValue"
-    | "GetCompareExchangeValue"
     | "SubscriptionResponsePayload"
     | "SubscriptionRevisionsResponsePayload";
 
@@ -45,17 +44,6 @@ export function getTransformJsonKeysProfile(
             };
         }
 
-        if (profile === "GetCompareExchangeValue") {
-            if (!conventions) {
-                throwError("InvalidArgumentException", "Document conventions are required for this profile.");
-            }
-
-            return {
-                getCurrentTransform:
-                    buildEntityKeysTransformForGetCompareExchangeValue(conventions.entityFieldNameConvention)
-            };
-        }
-
         if (profile === "SubscriptionResponsePayload") {
             if (!conventions) {
                 throwError("InvalidArgumentException", "Document conventions are required for this profile.");
@@ -176,24 +164,6 @@ function buildEntityKeysTransformForPutCompareExchangeValue(entityCasingConventi
     };
 }
 
-function buildEntityKeysTransformForGetCompareExchangeValue(entityCasingConvention) {
-    return function getCompareExchangeValueTransform(key, stack) {
-        const len = stack.length;
-
-        if (stack[0] === "Results") {
-            if (stack[2] === "Value" && stack[3] === "@metadata") {
-                return handleMetadataJsonKeys(key, stack, len, 4);
-            }
-        }
-
-        if (len <= 4) {
-            return "camel";
-        }
-
-        return entityCasingConvention;
-    };
-}
-
 function buildEntityKeysTransformForSubscriptionResponsePayload(entityCasingConvention) {
     return function entityKeysTransform(key, stack) {
         const len = stack.length;
diff --git a/test/Documents/CustomKeyCaseConventionsTests.ts b/test/Documents/CustomKeyCaseConventionsTests.ts
@@ -273,7 +273,7 @@ describe("With custom key case conventions set", function () {
 
             const cmp = await session.advanced.clusterTransaction.getCompareExchangeValue<typeof pascalCasedCmpXchObj>("cmp/1");
             // we don't transform casing for cmpXchValues
-            assert.strictEqual(cmp.value.Email, pascalCasedCmpXchObj.Email);
+            assert.strictEqual(cmp.value.email, pascalCasedCmpXchObj.Email);
 
             assertThat(session.advanced.numberOfRequests)
                 .isEqualTo(1);
diff --git a/test/Ported/Core/Streaming/QueryStreaming.ts b/test/Ported/Core/Streaming/QueryStreaming.ts
@@ -10,9 +10,10 @@ import * as assert from "assert";
 import { User } from "../../../Assets/Entities";
 import * as StreamUtil from "../../../../src/Utility/StreamUtil";
 import { CONSTANTS } from "../../../../src/Constants";
-import { parseJsonVerbose } from "../../../Utils/Json";
+import { parseJsonStreamVerbose, parseJsonVerbose } from "../../../Utils/Json";
 import { getStringWritable } from "../../../Utils/Streams";
 import { assertThat } from "../../../Utils/AssertExtensions";
+import * as Parser from "stream-json/Parser";
 
 describe("query streaming", function () {
 
@@ -310,14 +311,9 @@ describe("query streaming", function () {
 
             const result: string = targetStream["string"];
             assert.ok(result);
-            const json = parseJsonVerbose(result);
+
+            const json = await parseJsonStreamVerbose(result);
             assert.ok(json);
-            const res = json.results;
-            assert.ok(res);
-            assert.strictEqual(json.indexName, "Users/ByName");
-            assert.ok(json.indexTimestamp);
-            assert.strictEqual(json.isStale, false);
-            assert.ok("resultEtag" in json);
         }
     });
 });
diff --git a/test/Utils/Json.ts b/test/Utils/Json.ts
@@ -1,3 +1,5 @@
+import * as Parser from "stream-json/Parser";
+
 export function parseJsonVerbose(jsonString: string) {
     if (!jsonString) {
         throw new Error("JSON cannot be empty.");
@@ -11,3 +13,26 @@ export function parseJsonVerbose(jsonString: string) {
         throw err;
     }
 }
+
+
+export async function parseJsonStreamVerbose(jsonString: string) {
+    if (!jsonString) {
+        throw new Error("JSON cannot be empty.");
+    }
+
+    const parser = new Parser({ jsonStreaming: true, streamValues: false });
+    parser.push(jsonString);
+    parser.push(null);
+
+    return new Promise<any>((resolve) => {
+        const items = [];
+
+        parser.on("data", x => {
+            items.push(x);
+        });
+
+        parser.on("end", () => {
+            resolve(items);
+        });
+    });
+}

Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ export class RavenCommandResponsePipeline<TStreamResult> extends EventEmitter {`
`93`	`93`	`}`
`94`	`94`	`});`
`95`	`95`
`96`		`- transforms.push(extractItemTransform);`
	`96`	`+ transforms.unshift(extractItemTransform);`
`97`	`97`
`98`	`98`	`this._opts.jsonlAsync = {`
`99`	`99`	`transforms`