sync with java: 4c961f143020b34aec8edd04963b599b3475c365

Author: ml054

src/Documents/BulkInsertOperation.ts

@@ -28,6 +28,7 @@ import { TimeSeriesOperations } from "./TimeSeries/TimeSeriesOperations";
 import { TimeSeriesValuesHelper } from "./Session/TimeSeries/TimeSeriesValuesHelper";
 
 export class BulkInsertOperation {
+    private _options: BulkInsertOptions;
     private readonly _generateEntityIdOnTheClient: GenerateEntityIdOnTheClient;
 
     private readonly _requestExecutor: RequestExecutor;
@@ -45,6 +46,7 @@ export class BulkInsertOperation {
     private readonly _timeSeriesBatchSize: number;
 
     private _concurrentCheck: number = 0;
+    private _isInitialWrite: boolean = true;
 
     private _bulkInsertAborted: Promise<void>;
     private _abortReject: Function;
@@ -53,12 +55,15 @@ export class BulkInsertOperation {
     private _requestBodyStream: stream.PassThrough;
     private _pipelineFinished: Promise<void>;
 
-    public constructor(database: string, store: IDocumentStore) {
+    public constructor(database: string, store: IDocumentStore, options?: BulkInsertOptions) {
         this._conventions = store.conventions;
         if (StringUtil.isNullOrEmpty(database)) {
             this._throwNoDatabase();
         }
         this._requestExecutor = store.getRequestExecutor(database);
+        this._useCompression = options ? options.useCompression : false;
+
+        this._options = options ?? {};
 
         this._timeSeriesBatchSize = this._conventions.bulkInsert.timeSeriesBatchSize;
 
@@ -310,7 +315,7 @@ export class BulkInsertOperation {
 
             this._requestBodyStream = new stream.PassThrough();
             const bulkCommand =
-                new BulkInsertCommand(this._operationId, this._requestBodyStream, this._nodeTag);
+                new BulkInsertCommand(this._operationId, this._requestBodyStream, this._nodeTag, this._options.skipOverwriteIfUnchanged);
             bulkCommand.useCompression = this._useCompression;
 
             const bulkCommandPromise = this._requestExecutor.execute(bulkCommand);
@@ -787,19 +792,25 @@ export class BulkInsertCommand extends RavenCommand<void> {
     }
 
     private readonly _stream: stream.Readable;
+    private _skipOverwriteIfUnchanged: boolean;
     private readonly _id: number;
     public useCompression: boolean;
 
-    public constructor(id: number, stream: stream.Readable, nodeTag: string) {
+    public constructor(id: number, stream: stream.Readable, nodeTag: string, skipOverwriteIfUnchanged: boolean) {
         super();
 
         this._stream = stream;
         this._id = id;
         this._selectedNodeTag = nodeTag;
+        this._skipOverwriteIfUnchanged = skipOverwriteIfUnchanged;
     }
 
     public createRequest(node: ServerNode): HttpRequestParameters {
-        const uri = node.url + "/databases/" + node.database + "/bulk_insert?id=" + this._id;
+        const uri = node.url
+            + "/databases/" + node.database
+            + "/bulk_insert?id=" + this._id
+            + "&skipOverwriteIfUnchanged=" + (this._skipOverwriteIfUnchanged ? "true" : "false");
+
         const headers = this._headers().typeAppJson().build();
         // TODO: useCompression ? new GzipCompressingEntity(_stream) : _stream);
         return { 

src/Documents/Commands/ConditionalGetDocumentsCommand.ts

@@ -8,6 +8,7 @@ import { DocumentConventions } from "../Conventions/DocumentConventions";
 import { readToEnd, stringToReadable } from "../../Utility/StreamUtil";
 import { RavenCommandResponsePipeline } from "../../Http/RavenCommandResponsePipeline";
 import { ObjectUtil } from "../../Utility/ObjectUtil";
+import { CONSTANTS, HEADERS } from "../../Constants";
 
 export class ConditionalGetDocumentsCommand extends RavenCommand<ConditionalGetResult> {
 
@@ -31,7 +32,7 @@ export class ConditionalGetDocumentsCommand extends RavenCommand<ConditionalGetR
             uri,
             method: "GET",
             headers: {
-                "If-None-Match": `"${this._changeVector}"`
+                [HEADERS.IF_NONE_MATCH]: `"${this._changeVector}"`
             }
         }
     }
@@ -101,4 +102,4 @@ export class ConditionalGetDocumentsCommand extends RavenCommand<ConditionalGetR
 export interface ConditionalGetResult {
     results: any[];
     changeVector: string;
-}
+}

src/Documents/Commands/GetDocumentsCommand.ts

@@ -43,6 +43,8 @@ export interface GetDocumentsByIdsCommandOptions
     includes?: string[];
     metadataOnly?: boolean;
     timeSeriesIncludes?: AbstractTimeSeriesRange[];
+    revisionsIncludesByChangeVector?: string[];
+    revisionIncludeByDateTimeBefore?: Date;
     compareExchangeValueIncludes?: string[];
 }
 
@@ -61,6 +63,7 @@ export interface GetDocumentsResult {
     includes: IRavenObject;
     results: any[];
     counterIncludes: IRavenObject;
+    revisionIncludes: any[];
     timeSeriesIncludes: IRavenObject;
     compareExchangeValueIncludes: IRavenObject;
     nextPageStart: number;
@@ -77,6 +80,8 @@ export class GetDocumentsCommand extends RavenCommand<GetDocumentsResult> {
     private _includeAllCounters: boolean;
 
     private _timeSeriesIncludes: AbstractTimeSeriesRange[];
+    private _revisionsIncludeByChangeVector: string[];
+    private _revisionsIncludeByDateTime: Date;
     private _compareExchangeValueIncludes: string[];
 
     private readonly _metadataOnly: boolean;
@@ -114,6 +119,8 @@ export class GetDocumentsCommand extends RavenCommand<GetDocumentsResult> {
             this._metadataOnly = opts.metadataOnly;
             this._timeSeriesIncludes = opts.timeSeriesIncludes;
             this._compareExchangeValueIncludes = opts.compareExchangeValueIncludes;
+            this._revisionsIncludeByDateTime = opts.revisionIncludeByDateTimeBefore;
+            this._revisionsIncludeByChangeVector = opts.revisionsIncludesByChangeVector;
         } else if (opts.hasOwnProperty("start") && opts.hasOwnProperty("pageSize")) {
             opts = opts as GetDocumentsStartingWithOptions;
             this._start = opts.start;
@@ -224,6 +231,16 @@ export class GetDocumentsCommand extends RavenCommand<GetDocumentsResult> {
             }
         }
 
+        if (this._revisionsIncludeByChangeVector) {
+            for (const changeVector of this._revisionsIncludeByChangeVector) {
+                query += "&revisions=" + this._urlEncode(changeVector);
+            }
+        }
+
+        if (this._revisionsIncludeByDateTime) {
+            query += "&revisionsBefore=" + DateUtil.utc.stringify(this._revisionsIncludeByDateTime);
+        }
+
         if (this._compareExchangeValueIncludes) {
             for (const compareExchangeValue of this._compareExchangeValueIncludes) {
                 query += "&cmpxchg=" + this._urlEncode(compareExchangeValue);

src/Documents/Commands/GetRevisionsCommand.ts

@@ -59,6 +59,18 @@ export class GetRevisionsCommand extends RavenCommand<IRavenArrayResult> {
         this._conventions = conventions;
     }
 
+    public get id(): string {
+        return this._id;
+    }
+
+    public get before(): Date {
+        return this._before;
+    }
+
+    public get changeVector(): string {
+        return this._changeVector;
+    }
+
     public get changeVectors() {
         return this._changeVectors;
     }

src/Documents/Commands/HeadAttachmentCommand.ts

@@ -7,6 +7,7 @@ import { ServerNode } from "../../Http/ServerNode";
 import { StatusCodes } from "./../../Http/StatusCode";
 import * as stream from "readable-stream";
 import { getRequiredEtagHeader } from "../../Utility/HttpUtil";
+import { HEADERS } from "../../Constants";
 
 export class HeadAttachmentCommand extends RavenCommand<string> {
 
@@ -41,12 +42,14 @@ export class HeadAttachmentCommand extends RavenCommand<string> {
             + "/attachments?id=" + encodeURIComponent(this._documentId)
             + "&name=" + encodeURIComponent(this._name);
 
-        const req = {
+        const req: HttpRequestParameters = {
             method: "HEAD",
             uri
         };
 
-        this._addChangeVectorIfNotNull(this._changeVector, req);
+        if (this._changeVector) {
+            req.headers[HEADERS.IF_NONE_MATCH] = `"${this._changeVector}"`;
+        }
 
         return req;
     }

src/Documents/Commands/HeadDocumentCommand.ts

@@ -6,6 +6,7 @@ import { HttpCache } from "../../Http/HttpCache";
 import { getRequiredEtagHeader } from "../../Utility/HttpUtil";
 import { ServerNode } from "../../Http/ServerNode";
 import * as stream from "readable-stream";
+import { HEADERS } from "../../Constants";
 
 export class HeadDocumentCommand extends RavenCommand<string> {
 
@@ -34,7 +35,7 @@ export class HeadDocumentCommand extends RavenCommand<string> {
         const headers = this._headers()
             .typeAppJson();
         if (this._changeVector) {
-            headers.with("If-None-Match", this._changeVector);
+            headers.with(HEADERS.IF_NONE_MATCH, this._changeVector);
         }
 
         return {

src/Documents/Commands/MultiGet/MultiGetCommand.ts

@@ -171,7 +171,7 @@ export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDis
             for (let i = 0; i < responses.length; i++) {
                 const res = responses[i];
                 const command = this._commands[i];
-                this._maybeSetCache(res, command);
+                this._maybeSetCache(res, command, i);
 
                 if (this._cached && res.statusCode === StatusCodes.NotModified) {
                     const clonedResponse = new GetResponse();
@@ -191,8 +191,12 @@ export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDis
         }
     }
 
-    private _maybeSetCache(getResponse: GetResponse, command: GetRequest): void {
+    private _maybeSetCache(getResponse: GetResponse, command: GetRequest, cachedIndex: number): void {
         if (getResponse.statusCode === StatusCodes.NotModified) {
+            // if not modified - update age
+            if (this._cached) {
+                this._cached.values[cachedIndex][0].notModified();
+            }
             return;
         }
 
@@ -220,11 +224,24 @@ export class MultiGetCommand extends RavenCommand<GetResponse[]> implements IDis
     }
 
     public closeCache() {
+        //If _cached is not null - it means that the client approached with this multitask request to node and the request failed.
+        //and now client tries to send it to another node.
         if (this._cached) {
             this._cached.dispose();
-        }
 
-        this._cached = null;
+            this._cached = null;
+
+            // The client sends the commands.
+            // Some of which could be saved in cache with a response
+            // that includes the change vector that received from the old fallen node.
+            // The client can't use those responses because their URLs are different
+            // (include the IP and port of the old node), because of that the client
+            // needs to get those docs again from the new node.
+
+            for (let command of this._commands) {
+                delete command.headers[HEADERS.IF_NONE_MATCH];
+            }
+        }
     }
 }
 
@@ -245,4 +262,4 @@ class Cached implements IDisposable {
 
         this.values = null;
     }
-}
+}

src/Documents/DocumentStore.ts

@@ -12,13 +12,14 @@ import { IDocumentSession } from "./Session/IDocumentSession";
 import { SessionOptions } from "./Session/SessionOptions";
 import { DocumentSession } from "./Session/DocumentSession";
 import { IAuthOptions } from "../Auth/AuthOptions";
-import { BulkInsertOperation } from "./BulkInsertOperation";
+import { BulkInsertOperation, BulkInsertOptions } from "./BulkInsertOperation";
 import { IDatabaseChanges } from "./Changes/IDatabaseChanges";
 import { DatabaseChanges } from "./Changes/DatabaseChanges";
 import { DatabaseSmuggler } from "./Smuggler/DatabaseSmuggler";
 import { DatabaseChangesOptions } from "./Changes/DatabaseChangesOptions";
 import { IDisposable } from "../Types/Contracts";
 import { MultiDatabaseHiLoIdGenerator } from "./Identity/MultiDatabaseHiLoIdGenerator";
+import { TypeUtil } from "../Utility/TypeUtil";
 
 const log = getLogger({ module: "DocumentStore" });
 
@@ -84,6 +85,10 @@ export class DocumentStore extends DocumentStoreBase {
         this._identifier = identifier;
     }
 
+    public get hiLoIdGenerator() {
+        return this._multiDbHiLo;
+    }
+
     /**
      * Disposes the document store
      */
@@ -420,9 +425,14 @@ export class DocumentStore extends DocumentStoreBase {
 
     public bulkInsert(): BulkInsertOperation;
     public bulkInsert(database: string): BulkInsertOperation;
-    public bulkInsert(database?: string): BulkInsertOperation {
+    public bulkInsert(options: BulkInsertOptions): BulkInsertOperation;
+    public bulkInsert(database: string, options: BulkInsertOptions): BulkInsertOperation;
+    public bulkInsert(databaseOrOptions?: string | BulkInsertOptions, optionalOptions?: BulkInsertOptions): BulkInsertOperation {
         this.assertInitialized();
 
-        return new BulkInsertOperation(this.getEffectiveDatabase(database), this);
+        const database = TypeUtil.isString(databaseOrOptions) ? this.getEffectiveDatabase(databaseOrOptions) : this.getEffectiveDatabase(null);
+        const options: BulkInsertOptions = TypeUtil.isString(databaseOrOptions) ? optionalOptions : databaseOrOptions;
+
+        return new BulkInsertOperation(database, this, options);
     }
 }

src/Documents/DocumentStoreBase.ts

@@ -25,7 +25,7 @@ import { DocumentConventions } from "./Conventions/DocumentConventions";
 import { RequestExecutor } from "../Http/RequestExecutor";
 import { IndexCreation } from "../Documents/Indexes/IndexCreation";
 import { PutIndexesOperation } from "./Operations/Indexes/PutIndexesOperation";
-import { BulkInsertOperation } from "./BulkInsertOperation";
+import { BulkInsertOperation, BulkInsertOptions } from "./BulkInsertOperation";
 import { IDatabaseChanges } from "./Changes/IDatabaseChanges";
 import { DocumentSubscriptions } from "./Subscriptions/DocumentSubscriptions";
 import { DocumentStore } from "./DocumentStore";
@@ -37,6 +37,7 @@ import { IDisposable } from "../Types/Contracts";
 import { TimeSeriesOperations } from "./TimeSeries/TimeSeriesOperations";
 import { IAbstractIndexCreationTask } from "./Indexes/IAbstractIndexCreationTask";
 import { StringUtil } from "../Utility/StringUtil";
+import { IHiLoIdGenerator } from "./Identity/IHiLoIdGenerator";
 
 export abstract class DocumentStoreBase
     extends EventEmitter
@@ -70,6 +71,8 @@ export abstract class DocumentStoreBase
 
     public abstract identifier: string;
 
+    public abstract hiLoIdGenerator: IHiLoIdGenerator;
+
     public abstract initialize(): IDocumentStore;
 
     public abstract openSession(): IDocumentSession;
@@ -153,7 +156,10 @@ export abstract class DocumentStoreBase
 
     private _authOptions: IAuthOptions;
 
-    public abstract bulkInsert(database?: string): BulkInsertOperation;
+    public abstract bulkInsert(): BulkInsertOperation;
+    public abstract bulkInsert(database: string): BulkInsertOperation;
+    public abstract bulkInsert(database: string, options: BulkInsertOptions): BulkInsertOperation;
+    public abstract bulkInsert(options: BulkInsertOptions): BulkInsertOperation;
 
     private readonly _subscriptions: DocumentSubscriptions;