# HG changeset patch
# User cin
# Date 2019-06-17 05:51:35
# Node ID a193ba786ffc333601ceeba3df872e5e036d4ce9
# Parent  682bf9cf6f0c87c91c12184b8b60204983b41527

Added safe.delay
more tests

diff --git a/build.md b/build.md
--- a/build.md
+++ b/build.md
@@ -65,4 +65,4 @@
 
 ```shell
 ./gradlew test pack -Pjsmodule=commonjs -Ptarget=es2017
-```
\ No newline at end of file
+```
diff --git a/src/main/ts/Cancellation.ts b/src/main/ts/Cancellation.ts
--- a/src/main/ts/Cancellation.ts
+++ b/src/main/ts/Cancellation.ts
@@ -1,16 +1,11 @@
 import { ICancellation, IDestroyable } from "./interfaces";
-import { argumentNotNull } from "./safe";
-
-const destroyed = {
-    destroy() {
-    }
-};
+import { argumentNotNull, destroyed } from "./safe";
 
 export class Cancellation implements ICancellation {
     private _reason: any;
-    private _cbs: Array<(e) => void>;
+    private _cbs: Array<(e: any) => void>;
 
-    constructor(action: (cancel: (e) => void) => void) {
+    constructor(action: (cancel: (e?: any) => void) => void) {
         argumentNotNull(action, "action");
 
         action(this._cancel.bind(this));
diff --git a/src/main/ts/Observable.ts b/src/main/ts/Observable.ts
--- a/src/main/ts/Observable.ts
+++ b/src/main/ts/Observable.ts
@@ -1,22 +1,17 @@
-import { IObservable, IDestroyable, ICancellation } from "./interfaces";
+import { IObservable, IDestroyable, ICancellation, IObserver } from "./interfaces";
 import { Cancellation } from "./Cancellation";
-import { argumentNotNull } from "./safe";
+import { argumentNotNull, destroyed } from "./safe";
 
 type Handler<T> = (x: T) => void;
 
 type Initializer<T> = (notify: Handler<T>, error?: (e: any) => void, complete?: () => void) => void;
 
-// TODO: think about to move this interfaces.ts and make it public
-interface IObserver<T> {
-    next(event: T): void;
+const noop = () => { };
 
-    error(e: any): void;
-
-    complete(): void;
+function isObserver(val: any): val is IObserver<any> {
+    return val && (typeof val.next === "function");
 }
 
-const noop = () => { };
-
 export class Observable<T> implements IObservable<T> {
     private _once = new Array<IObserver<T>>();
 
@@ -65,6 +60,35 @@ export class Observable<T> implements IO
         return observer;
     }
 
+    subscribe(next: IObserver<T> | Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
+        if (isObserver(next)) {
+            const me = this;
+            const subscription = {
+                destroy() {
+                    me._removeObserver(next);
+                }
+            };
+            this._addObserver(next);
+            return subscription;
+        } else if (next) {
+            const observer = {
+                next,
+                error,
+                complete
+            };
+            const me = this;
+            const subscription = {
+                destroy() {
+                    me._removeObserver(observer);
+                }
+            };
+            this._addObserver(observer);
+            return subscription;
+        } else {
+            return destroyed;
+        }
+    }
+
     private _addObserver(observer: IObserver<T>) {
         if (this._complete) {
             try {
@@ -86,7 +110,7 @@ export class Observable<T> implements IO
      *
      * @param ct a cancellation token
      */
-    next(ct: ICancellation = Cancellation.none): Promise<T> {
+    next(ct: ICancellation = Cancellation.none) {
         return new Promise<T>((resolve, reject) => {
             const observer: IObserver<T> = {
                 next: resolve,
diff --git a/src/main/ts/di/Configuration.ts b/src/main/ts/di/Configuration.ts
--- a/src/main/ts/di/Configuration.ts
+++ b/src/main/ts/di/Configuration.ts
@@ -62,7 +62,7 @@ export class Configuration {
     _require: ModuleResolver;
 
     constructor(container: Container) {
-        argumentNotNull(container, container);
+        argumentNotNull(container, "container");
         this._container = container;
         this._path = [];
     }
diff --git a/src/main/ts/interfaces.ts b/src/main/ts/interfaces.ts
--- a/src/main/ts/interfaces.ts
+++ b/src/main/ts/interfaces.ts
@@ -91,3 +91,11 @@ export interface IObservable<T> {
     on(next: (x: T) => void, error?: (e: any) => void, complete?: () => void): IDestroyable;
     next(ct?: ICancellation): Promise<T>;
 }
+
+export interface IObserver<T> {
+    next(event: T): void;
+
+    error(e: any): void;
+
+    complete(): void;
+}
diff --git a/src/main/ts/safe.ts b/src/main/ts/safe.ts
--- a/src/main/ts/safe.ts
+++ b/src/main/ts/safe.ts
@@ -1,4 +1,5 @@
 import { ICancellable, Constructor } from "./interfaces";
+import { Cancellation } from "./Cancellation";
 
 let _nextOid = 0;
 const _oid = typeof Symbol === "function" ?
@@ -246,6 +247,24 @@ export function delegate<T, K extends ke
     };
 }
 
+export function delay(timeMs: number, ct = Cancellation.none) {
+    return new Promise((resolve, reject) => {
+        if (ct.isRequested()) {
+            ct.register(reject);
+        } else {
+            const h = ct.register(e => {
+                clearTimeout(id);
+                reject(e);
+                // we don't nedd to unregister h, since ct is already disposed
+            });
+            const id = setTimeout(() => {
+                h.destroy();
+                resolve();
+            }, timeMs);
+        }
+    });
+}
+
 /**
  * Для каждого элемента массива вызывает указанную функцию и сохраняет
  * возвращенное значение в массиве результатов.
@@ -381,8 +400,8 @@ export function firstWhere<T>(
 export function firstWhere<T>(
     sequence: ArrayLike<T> | PromiseLike<ArrayLike<T>>,
     predicate?: (x: T) => boolean,
-    cb?: (x: T) => void,
-    err?: (x: Error) => void
+    cb?: (x: T) => any,
+    err?: (x: Error) => any
 ) {
     if (isPromise(sequence)) {
         return sequence.then(res => firstWhere(res, predicate, cb, err));
@@ -394,7 +413,7 @@ export function firstWhere<T>(
                 throw new Error("The sequence is empty");
         } else {
             if (!predicate) {
-                return cb ? cb(sequence[0]) : sequence[0];
+                return cb ? cb(sequence[0]) && void (0) : sequence[0];
             } else {
                 for (let i = 0; i < sequence.length; i++) {
                     const v = sequence[i];
@@ -426,3 +445,12 @@ export function destroy(d: any) {
  */
 export function nowait(p: Promise<any>) {
 }
+
+/** represents already destroyed object.
+ */
+export const destroyed = {
+    /** Calling to this method doesn't affect anything, noop.
+     */
+    destroy() {
+    }
+};
diff --git a/src/test/ts/SafeTests.ts b/src/test/ts/SafeTests.ts
new file mode 100644
--- /dev/null
+++ b/src/test/ts/SafeTests.ts
@@ -0,0 +1,86 @@
+import tape = require("tape");
+import { delay } from "./TestTraits";
+import { Cancellation } from "@implab/core/Cancellation";
+import { first, isPromise } from "@implab/core/safe";
+
+tape("await delay test", async t => {
+    // schedule delay
+    let resolved = false;
+    let res = delay(0).then(() => resolved = true);
+
+    t.false(resolved, "the delay should be async");
+
+    await res;
+    t.pass("await delay");
+
+    // create cancellation token
+    let cancel: (e?: any) => void;
+    const ct = new Cancellation(c => cancel = c);
+
+    // schedule delay
+    resolved = false;
+    res = delay(0, ct).then(() => resolved = true);
+
+    t.false(resolved, "created delay with ct");
+
+    // cancel
+    cancel();
+
+    try {
+        await res;
+        t.fail("the delay should fail when it is cancelled");
+    } catch {
+        t.pass("the delay is cancelled");
+    }
+
+    let died = false;
+
+    // try schedule delay after the cancellation is requested
+    res = delay(0, ct).then(x => true, () => died = true);
+
+    t.false(died, "The delay should be scheduled even if the cancellation is requested");
+
+    await res;
+    t.true(died, "the delay should fail when cancelled");
+
+    t.end();
+});
+
+tape("sequemce test", async t => {
+    const sequence = ["a", "b", "c"];
+    const empty = [];
+
+    // synchronous tests
+    t.equals(first(sequence), "a", "Should return the first element");
+
+    let v: string;
+    let e: Error;
+    first(sequence, x => v = x);
+    t.equal(v, "a", "The callback should be called for the first element");
+
+    t.throws(() => {
+        first(empty);
+    }, "Should throw when the sequence is empty");
+
+    t.throws(() => {
+        first(empty, x => v = x);
+    }, "Should throw when the sequence is empty");
+
+    first(empty, null, x => e = x);
+    t.true(e, "The errorback should be called for the empty sequence");
+
+    // async tests
+    const asyncSequence = Promise.resolve(sequence);
+    const asyncEmptySequence = Promise.resolve(empty);
+
+    const promise = first(asyncSequence);
+    t.true(isPromise(promise), "Should return promise");
+
+    v = await promise;
+    t.equal(v, "a", "Should return the first element");
+
+    v = await new Promise(resolve => first(asyncSequence, resolve));
+    t.equal(v, "a", "The callback should be called for the first element");
+
+    t.end();
+});
diff --git a/src/testAmd/js/plan.js b/src/testAmd/js/plan.js
--- a/src/testAmd/js/plan.js
+++ b/src/testAmd/js/plan.js
@@ -4,5 +4,6 @@ define([
     "./TraceSourceTests",
     "./CancellationTests",
     "./ObservableTests",
-    "./ContainerTests"
+    "./ContainerTests",
+    "./SafeTests"
 ]);
\ No newline at end of file