Merge pull request #157 from gregolsky/subscriptions

Subscriptions & bug fixes

Author: gregolsky

package-lock.json

@@ -65,12 +65,10 @@
     "typescript": "^2.9.2"
   },
   "dependencies": {
-    "@types/change-case": "^2.3.1",
     "@types/pluralize": "0.0.27",
     "@types/qs": "^6.5.1",
     "@types/request": "^2.0.3",
     "@types/semaphore": "^1.1.0",
-    "@types/through2": "^2.0.33",
     "@types/uuid": "^2.0.29",
     "@types/verror": "^1.10.3",
     "bluebird": "^3.4.7",
@@ -79,7 +77,7 @@
     "moment": "^2.22.0",
     "pluralize": "^4.0.0",
     "qs": "^6.5.2",
-    "readable-stream": "^3.0.3",
+    "readable-stream": "^3.0.6",
     "request": "^2.88.0",
     "safe-memory-cache": "^1.5.1",
     "semaphore": "^1.1.0",

package.json

@@ -62,9 +62,11 @@ export abstract class Certificate implements ICertificate {
     }
 
     public toWebSocketOptions(): WebSocket.ClientOptions {
-        return {
-            passphrase: this._passphrase
-        };
+        if (this._passphrase) {
+            return { passphrase: this._passphrase };
+        }
+
+        return {};
     }
 }
 

src/Auth/Certificate.ts

@@ -0,0 +1,43 @@
+import { DocumentConventions, ServerNode } from "../..";
+import { RavenCommand } from "../../Http/RavenCommand";
+import { CreateSubscriptionResult } from "../Subscriptions/CreateSubscriptionResult";
+import { SubscriptionCreationOptions } from "../Subscriptions/SubscriptionCreationOptions";
+import { HttpRequestParameters } from "../../Primitives/Http";
+import * as stream from "readable-stream";
+
+export class CreateSubscriptionCommand extends RavenCommand<CreateSubscriptionResult> {
+    private readonly _conventions: DocumentConventions;
+    private readonly _options: SubscriptionCreationOptions;
+    private readonly _id: string;
+
+    public constructor(conventions: DocumentConventions, options: SubscriptionCreationOptions, id?: string) {
+        super();
+        this._conventions = conventions;
+        this._options = options;
+        this._id = id;
+    }
+
+    public createRequest(node: ServerNode): HttpRequestParameters {
+        let uri = node.url + "/databases/" + node.database + "/subscriptions";
+
+        if (this._id) {
+            uri += "?id=" + this._id;
+        }
+
+        const body = this._serializer.serialize(this._options);
+
+        return {
+            uri,
+            method: "PUT",
+            body
+        };
+    }
+
+    public async setResponseAsync(bodyStream: stream.Stream, fromCache: boolean): Promise<string> {
+        return this._parseResponseDefaultAsync(bodyStream);
+    }
+
+    public get isReadRequest() {
+        return false;
+    }
+}

src/Documents/Commands/CreateSubscriptionCommand.ts

@@ -0,0 +1,24 @@
+import { ServerNode } from "../..";
+import { RavenCommand } from "../../Http/RavenCommand";
+import { HttpRequestParameters } from "../../Primitives/Http";
+
+export class DeleteSubscriptionCommand extends RavenCommand<void> {
+    private readonly _name: string;
+
+    public constructor(name: string) {
+        super();
+        this._name = name;
+    }
+
+    public createRequest(node: ServerNode): HttpRequestParameters {
+        const uri = node.url + "/databases/" + node.database + "/subscriptions?taskName=" + this._name;
+        return {
+            uri,
+            method: "DELETE"
+        };
+    }
+
+    public get isReadRequest() {
+        return false;
+    }
+}

src/Documents/Commands/DeleteSubscriptionCommand.ts

@@ -0,0 +1,26 @@
+import { ServerNode } from "../..";
+import { RavenCommand } from "../../Http/RavenCommand";
+import { HttpRequestParameters } from "../../Primitives/Http";
+
+export class DropSubscriptionConnectionCommand extends RavenCommand<void> {
+
+    private readonly _name: string;
+
+    public constructor(name: string) {
+        super();
+        this._name = name;
+    }
+
+    public createRequest(node: ServerNode): HttpRequestParameters {
+        const uri = node.url + "/databases/" + node.database + "/subscriptions/drop?name=" + this._name;
+
+        return {
+            method: "POST",
+            uri
+        };
+    }
+
+    public get isReadRequest() {
+        return false;
+    }
+}

src/Documents/Commands/DropSubscriptionConnectionCommand.ts

@@ -0,0 +1,31 @@
+import { ServerNode } from "../..";
+import { RavenCommand } from "../../Http/RavenCommand";
+import { SubscriptionState } from "../Subscriptions/SubscriptionState";
+import { HttpRequestParameters } from "../../Primitives/Http";
+import * as stream from "readable-stream";
+
+export class GetSubscriptionStateCommand extends RavenCommand<SubscriptionState> {
+
+    private readonly _subscriptionName: string;
+
+    public constructor(subscriptionName: string) {
+        super();
+        this._subscriptionName = subscriptionName;
+    }
+
+    public createRequest(node: ServerNode): HttpRequestParameters {
+        const uri = node.url + "/databases/" + node.database + "/subscriptions/state?name=" + this._subscriptionName;
+
+        return {
+            uri
+        };
+    }
+
+    public async setResponseAsync(bodyStream: stream.Stream, fromCache: boolean): Promise<string> {
+        return this._parseResponseDefaultAsync(bodyStream);
+    }
+
+    public get isReadRequest() {
+        return true;
+    }
+}

src/Documents/Commands/GetSubscriptionStateCommand.ts

@@ -0,0 +1,53 @@
+import { ServerNode } from "../..";
+import { RavenCommand } from "../../Http/RavenCommand";
+import { SubscriptionState } from "../Subscriptions/SubscriptionState";
+import { HttpRequestParameters } from "../../Primitives/Http";
+import * as stream from "readable-stream";
+
+export class GetSubscriptionsCommand extends RavenCommand<SubscriptionState[]> {
+
+    private readonly _start: number;
+    private readonly _pageSize: number;
+
+    public constructor(start: number, pageSize: number) {
+        super();
+
+        this._start = start;
+        this._pageSize = pageSize;
+    }
+
+    public createRequest(node: ServerNode): HttpRequestParameters {
+        const uri = node.url + "/databases/" + node.database
+            + "/subscriptions?start=" + this._start + "&pageSize=" + this._pageSize;
+
+        return {
+            uri
+        };
+    }
+
+    public async setResponseAsync(bodyStream: stream.Stream, fromCache: boolean): Promise<string> {
+        if (!bodyStream) {
+            this.result = null;
+            return;
+        }
+
+        let body: string = null;
+        await this._defaultPipeline(_ => body = _)
+            .process(bodyStream)
+            .then(data => {
+                const results = data["results"] as SubscriptionState[];
+                if (!results) {
+                    this._throwInvalidResponse();
+                    return;
+                }
+
+                this.result = results;
+            });
+
+        return body;
+    }
+
+    public get isReadRequest() {
+        return true;
+    }
+}

src/Documents/Commands/GetSubscriptionsCommand.ts

@@ -110,7 +110,7 @@ export class DocumentStore extends DocumentStoreBase {
             })
             .then(() => {
                 this._disposed = true;
-                // TBD: Subscriptions?.Dispose();
+                this.subscriptions.dispose();
 
                 return new BluebirdPromise((resolve, reject) => {
                     let listenersExecCallbacksCount = 0;

src/Documents/DocumentStore.ts

@@ -22,6 +22,8 @@ import { IndexCreation } from "../Documents/Indexes/IndexCreation";
 import { PutIndexesOperation } from "./Operations/Indexes/PutIndexesOperation";
 import { BulkInsertOperation } from "./BulkInsertOperation";
 import { IDatabaseChanges } from "./Changes/IDatabaseChanges";
+import { DocumentSubscriptions } from "./Subscriptions/DocumentSubscriptions";
+import { DocumentStore } from "./DocumentStore";
 
 export abstract class DocumentStoreBase
     extends EventEmitter
@@ -34,7 +36,7 @@ export abstract class DocumentStoreBase
 
     protected constructor() {
         super();
-        // TBD: Subscriptions = new DocumentSubscriptions(this);
+        this._subscriptions = new DocumentSubscriptions(this as any as DocumentStore);
     }
 
     public abstract dispose(): void;
@@ -140,7 +142,11 @@ export abstract class DocumentStoreBase
 
     public abstract bulkInsert(database?: string): BulkInsertOperation;
 
-    // TBD: public IReliableSubscriptions Subscriptions { get; }
+    private readonly _subscriptions: DocumentSubscriptions;
+
+    public get subscriptions(): DocumentSubscriptions {
+        return this._subscriptions;
+    }
 
     protected _ensureNotDisposed(): void {
         if (this._disposed) {