# HG changeset patch
# User cin
# Date 2018-09-07 20:14:53
# Node ID d5a3d3ab9fd76db3e1f312a0b5e0c95857534e9a
# Parent  53e756f117f743e725defd541bac1fd8e76089b7

fixed format-compile bug while formatting strings with new line symbols.
added error and completion handlers to Observable
ConsoleTraceWriter is now destroyable

diff --git a/package-lock.json b/package-lock.json
--- a/package-lock.json
+++ b/package-lock.json
@@ -133,7 +133,7 @@
         },
         "minimist": {
           "version": "0.0.5",
-          "resolved": "http://registry.npmjs.org/minimist/-/minimist-0.0.5.tgz",
+          "resolved": "https://registry.npmjs.org/minimist/-/minimist-0.0.5.tgz",
           "integrity": "sha1-16oye87PUY+RBqxrjwA/o7zqhWY=",
           "dev": true
         },
@@ -319,9 +319,9 @@
       }
     },
     "requirejs": {
-      "version": "2.3.6",
-      "resolved": "https://registry.npmjs.org/requirejs/-/requirejs-2.3.6.tgz",
-      "integrity": "sha512-ipEzlWQe6RK3jkzikgCupiTbTvm4S0/CAU5GlgptkN5SO6F3u0UD0K18wy6ErDqiCyP4J4YYe1HuAShvsxePLg==",
+      "version": "2.3.5",
+      "resolved": "https://registry.npmjs.org/requirejs/-/requirejs-2.3.5.tgz",
+      "integrity": "sha512-svnO+aNcR/an9Dpi44C7KSAy5fFGLtmPbaaCeQaklUz8BQhS64tWWIIlvEA5jrWICzlO/X9KSzSeXFnZdBu8nw==",
       "dev": true
     },
     "resolve": {
@@ -413,9 +413,9 @@
       }
     },
     "typescript": {
-      "version": "3.0.3",
-      "resolved": "https://registry.npmjs.org/typescript/-/typescript-3.0.3.tgz",
-      "integrity": "sha512-kk80vLW9iGtjMnIv11qyxLqZm20UklzuR2tL0QAnDIygIUIemcZMxlMWudl9OOt76H3ntVzcTiddQ1/pAAJMYg==",
+      "version": "2.9.2",
+      "resolved": "https://registry.npmjs.org/typescript/-/typescript-2.9.2.tgz",
+      "integrity": "sha512-Gr4p6nFNaoufRIY4NMdpQRNmgxVIGMs4Fcu/ujdYk3nAZqk7supzBE9idmvfZIlH/Cuj//dvi+019qEue9lV0w==",
       "dev": true
     },
     "wrappy": {
diff --git a/src/js/text/format-compile.js b/src/js/text/format-compile.js
--- a/src/js/text/format-compile.js
+++ b/src/js/text/format-compile.js
@@ -18,13 +18,13 @@ define(
         var espaceString = function(s) {
             if (!s)
                 return s;
-            return "'" + s.replace(/('|\\)/g, "\\$1") + "'";
+            return "'" + s.replace(/('|\\)/g, "\\$1").replace("\n","\\n") + "'";
         };
 
         var encode = function(s) {
             if (!s)
                 return s;
-            return s.replace(/\\{|\\}|&|\\:/g, function(m) {
+            return s.replace(/\\{|\\}|&|\\:|\n/g, function(m) {
                 return map[m] || m;
             });
         };
diff --git a/src/ts/components/Observable.ts b/src/ts/components/Observable.ts
--- a/src/ts/components/Observable.ts
+++ b/src/ts/components/Observable.ts
@@ -4,79 +4,162 @@ import { argumentNotNull } from '../safe
 
 
 interface Handler<T> {
-    (x:T) : void
+    (x: T): void
 }
 
 interface Initializer<T> {
-    (notify: Handler<T>) : (() => void) | void;
+    (notify: Handler<T>, error?: (e: any) => void, complete?: () => void): (() => void) | void;
 }
 
+// TODO: think about to move this interfaces.ts and make it public
+interface IObserver<T> {
+    next(event: T): void
+
+    error(e: any): void
+
+    complete(): void
+}
 
 class Observable<T> implements IObservable<T>, IDestroyable {
-    private _once = new Array<Handler<T>>();
+    private _once = new Array<IObserver<T>>();
+
+    private _observers = new Array<IObserver<T>>();
 
-    private readonly _observers = new Array<Handler<T>>();
+    private _cleanup: (() => void) | void;
 
-    private readonly _cleanup : (() => void) | void;
+    private _complete: boolean
+
+    private _error: any
 
     constructor(func?: Initializer<T>) {
-        this._cleanup = func && func(this._notify.bind(this));
+        this._cleanup = func && func(
+            this._notifyNext.bind(this),
+            this._notifyError.bind(this),
+            this._notifyCompleted.bind(this)
+        );
     }
 
-    on(observer: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
-        argumentNotNull(observer, "observer");
-
-        this._observers.push(observer);
+    /**
+     * Registers handlers for the current observable object.
+     * 
+     * @param next the handler for events
+     * @param error the handler for a error
+     * @param complete the handler for a completion
+     * @returns {IDestroyable} the handler for the current subscription, this
+     *  handler can be used to unsubscribe from events.
+     * 
+     */
+    on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
+        argumentNotNull(next, "next");
 
         let me = this;
-        return {
+
+        let observer: IObserver<T> & IDestroyable = {
+            next: next,
+
+            error(e: any) {
+                if (error)
+                    error(e);
+            },
+
+            complete() {
+                if (complete)
+                    complete();
+            },
+
             destroy() {
-                me._removeObserver(observer);
+                me._removeObserver(this);
             }
         }
+
+        this._addObserver(observer);
+
+
+        return observer;
+    }
+
+    private _addObserver(observer: IObserver<T>) {
+        if (this._complete) {
+            try {
+                if (this._error)
+                    observer.error(this._error);
+                else
+                    observer.complete();
+            } catch (e) {
+                this.onObserverException(e);
+            }
+        } else {
+            this._observers.push(observer);
+        }
     }
 
+    /**
+     * Waits for the next event. This method can't be used to read messages
+     * as a sequence since it can skip some messages between calls.
+     * 
+     * @param ct a cancellation token
+     */
     next(ct: ICancellation = Cancellation.none): Promise<T> {
         return new Promise<T>((resolve, reject) => {
-            this._once.push(resolve);
-            if (ct.isSupported()) {
+            let observer: IObserver<T> = {
+                next: resolve,
+                error: reject,
+                complete: () => reject("No more events are available")
+            };
+
+            if (this._addOnce(observer) && ct.isSupported()) {
                 ct.register((e) => {
-                    this._removeOnce(resolve);
+                    this._removeOnce(observer);
                     reject(e);
                 });
             }
         });
     }
 
+    private _addOnce(observer: IObserver<T>) {
+        if (this._complete) {
+            try {
+                if (this._error)
+                    observer.error(this._error);
+                else
+                    observer.complete();
+            } catch (e) {
+                this.onObserverException(e);
+            }
+            return false;
+        }
+
+        this._once.push(observer);
+        return true;
+    }
+
     destroy() {
-        if(this._cleanup)
-            this._cleanup.call(null);
+        if (this._complete)
+            this._notifyCompleted();
+
+        let cleanup = this._cleanup;
+        if (cleanup) {
+            this._cleanup = null;
+            cleanup();
+        }
     }
 
     protected onObserverException(e: any) {
     }
 
-    private _removeOnce(d: Handler<T>) {
+    private _removeOnce(d: IObserver<T>) {
         let i = this._once.indexOf(d);
         if (i >= 0)
             this._once.splice(i);
     }
 
-    private _removeObserver(d: Handler<T>) {
+    private _removeObserver(d: IObserver<T>) {
         let i = this._observers.indexOf(d);
         if (i >= 0)
             this._observers.splice(i);
     }
 
-    protected _notify(evt: T) {
-        let guard = (observer: Handler<T>) => {
-            try {
-                observer(evt);
-            } catch (e) {
-                this.onObserverException(e);
-            }
-        }
-
+    private _notify(guard: (observer: IObserver<T>) => void) {
         if (this._once.length) {
             for (let i = 0; i < this._once.length; i++)
                 guard(this._once[i]);
@@ -86,6 +169,44 @@ class Observable<T> implements IObservab
         for (let i = 0; i < this._observers.length; i++)
             guard(this._observers[i]);
     }
+
+    protected _notifyNext(evt: T) {
+        let guard = (observer: IObserver<T>) => {
+            try {
+                observer.next(evt);
+            } catch (e) {
+                this.onObserverException(e);
+            }
+        }
+
+        this._notify(guard);
+    }
+
+    protected _notifyError(e: any) {
+        let guard = (observer: IObserver<T>) => {
+            try {
+                observer.error(e);
+            } catch (e) {
+                this.onObserverException(e);
+            }
+        }
+
+        this._notify(guard);
+        this._observers = [];
+    }
+
+    protected _notifyCompleted() {
+        let guard = (observer: IObserver<T>) => {
+            try {
+                observer.complete();
+            } catch (e) {
+                this.onObserverException(e);
+            }
+        }
+
+        this._notify(guard);
+        this._observers = [];
+    }
 }
 
 namespace Observable {
diff --git a/src/ts/log/TraceSource.ts b/src/ts/log/TraceSource.ts
--- a/src/ts/log/TraceSource.ts
+++ b/src/ts/log/TraceSource.ts
@@ -67,7 +67,7 @@ class TraceSource extends Observable<Tra
     }
 
     protected emit(level: number, arg: any) {
-        this._notify(new TraceEvent(this, level, arg));
+        this._notifyNext(new TraceEvent(this, level, arg));
     }
 
     isDebugEnabled() {
diff --git a/src/ts/log/writers/ConsoleWriter.ts b/src/ts/log/writers/ConsoleWriter.ts
--- a/src/ts/log/writers/ConsoleWriter.ts
+++ b/src/ts/log/writers/ConsoleWriter.ts
@@ -1,25 +1,31 @@
-import { IObservable } from "../../interfaces";
+import { IObservable, IDestroyable, ICancellation } from "../../interfaces";
 import * as TraceEvent from '../TraceEvent';
-import { ICancellation } from "../../interfaces";
 import { Cancellation } from "../../Cancellation";
 import * as TraceSource from "../TraceSource";
 
-class ConsoleWriter {
-    async write(source: IObservable<TraceEvent>, ct: ICancellation = Cancellation.none) {
-        let next;
-        while(next = await source.next(ct)) {
-            this._writeEvent(next);
+class ConsoleWriter implements IDestroyable {
+    readonly _subscriptions = new Array<IDestroyable>();
+
+    writeEvents(source: IObservable<TraceEvent>, ct: ICancellation = Cancellation.none) {
+        var subscription = source.on(this.writeEvent.bind(this));
+        if (ct.isSupported()) {
+            ct.register(subscription.destroy.bind(subscription));
+        }
+        this._subscriptions.push(subscription);
+    }
+
+    writeEvent(next: TraceEvent) {
+        if (next.level >= TraceSource.LogLevel) {
+            console.log(next.source.id.toString(), next.arg);
+        } else if(next.level >= TraceSource.WarnLevel) {
+            console.warn(next.source.id.toString(), next.arg);
+        } else {
+            console.error(next.source.id.toString(), next.arg);
         }
     }
 
-    private _writeEvent(next: TraceEvent) {
-        if (next.level >= TraceSource.LogLevel) {
-            console.log(next.source.toString(), next.arg);
-        } else if(next.level >= TraceSource.WarnLevel) {
-            console.warn(next.source.toString(), next.arg);
-        } else {
-            console.error(next.source.toString(), next.arg);
-        }
+    destroy() {
+        this._subscriptions.forEach(x => x.destroy());
     }
 }
 
diff --git a/test/ts/TestTraits.ts b/test/ts/TestTraits.ts
new file mode 100644
--- /dev/null
+++ b/test/ts/TestTraits.ts
@@ -0,0 +1,39 @@
+import { IObservable, ICancellation, IDestroyable } from "@implab/core/interfaces";
+import * as TraceEvent from '@implab/core/log/TraceEvent';
+import { Cancellation } from "@implab/core/Cancellation";
+import * as TraceSource from "@implab/core/log/TraceSource";
+import * as tape from 'tape';
+import { argumentNotNull } from "@implab/core/safe";
+
+export class TapeWriter implements IDestroyable {
+    readonly _tape: tape.Test
+
+    _subscriptions = new Array<IDestroyable>();
+
+    constructor(tape: tape.Test) {
+        argumentNotNull(tape, "tape");
+        this._tape = tape;
+    }
+
+    writeEvents(source: IObservable<TraceEvent>, ct: ICancellation = Cancellation.none) {
+        let subscription = source.on(this.writeEvent.bind(this));
+        if (ct.isSupported()) {
+            ct.register(subscription.destroy.bind(subscription));
+        }
+        this._subscriptions.push(subscription);
+    }
+
+    writeEvent(next: TraceEvent) {
+        if (next.level >= TraceSource.LogLevel) {
+            this._tape.comment("LOG " + next.arg);
+        } else if(next.level >= TraceSource.WarnLevel) {
+            this._tape.comment("WARN " + next.arg);
+        } else {
+            this._tape.comment("ERROR " + next.arg);
+        }
+    }
+
+    destroy() {
+        this._subscriptions.forEach(x => x.destroy());
+    }
+}
\ No newline at end of file
diff --git a/test/ts/TraceSourceTests.ts b/test/ts/TraceSourceTests.ts
--- a/test/ts/TraceSourceTests.ts
+++ b/test/ts/TraceSourceTests.ts
@@ -1,6 +1,7 @@
 import * as TraceSource from '@implab/core/log/TraceSource'
 import * as tape from 'tape';
 import * as ConsoleWriter from '@implab/core/log/writers/ConsoleWriter';
+import { TapeWriter } from './TestTraits';
 
 const sourceId = 'test/TraceSourceTests';
 
@@ -44,19 +45,26 @@ tape('trace event', t => {
     h.destroy();
 });
 
-tape('console writer', async t => {
-    let writer = new ConsoleWriter();
+tape('tape comment writer', async t => {
+    let writer = new TapeWriter(t);
+
+    TraceSource.on(ts => {
+        writer.writeEvents(ts);
+    });
 
     let trace = TraceSource.get(sourceId);
     trace.level = TraceSource.DebugLevel;
 
-    let p = writer.write(trace);
-
     trace.log("Hello, {0}!", 'World');
+    trace.log("Multi\n  line");
     trace.warn("Look at me!");
     trace.error("DIE!");
 
-    console.log("DONE");
+    writer.destroy();
+    
+    trace.log("You shouldn't see it!");
+
+    t.comment("DONE");
 
     t.end();
 });
\ No newline at end of file