# HG changeset patch
# User cin
# Date 2018-09-06 00:06:49
# Node ID 53e756f117f743e725defd541bac1fd8e76089b7
# Parent  745612edbd745ac2abe1680971fb5247c647a663

working on Observable

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
@@ -1,23 +1,29 @@
-import { IObservable, IObserver, IDestroyable, ICancellation } from '../interfaces';
+import { IObservable, IDestroyable, ICancellation } from '../interfaces';
 import { Cancellation } from '../Cancellation'
-import * as TraceSource from '../log/TraceSource'
 import { argumentNotNull } from '../safe';
 
-const trace = TraceSource.get('@implab/core/components/Observable');
-
 
-class Observable<T> implements IObservable<T> {
-    private _once = new Array<IObserver<T>>();
+interface Handler<T> {
+    (x:T) : void
+}
 
-    private readonly _observers = new Array<IObserver<T>>();
+interface Initializer<T> {
+    (notify: Handler<T>) : (() => void) | void;
+}
+
 
-    constructor(func: (notify: IObserver<T>) => void) {
-        argumentNotNull(func, "func");
+class Observable<T> implements IObservable<T>, IDestroyable {
+    private _once = new Array<Handler<T>>();
+
+    private readonly _observers = new Array<Handler<T>>();
 
-        func(this._notify.bind(this));
+    private readonly _cleanup : (() => void) | void;
+
+    constructor(func?: Initializer<T>) {
+        this._cleanup = func && func(this._notify.bind(this));
     }
 
-    on(observer: IObserver<T>): IDestroyable {
+    on(observer: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
         argumentNotNull(observer, "observer");
 
         this._observers.push(observer);
@@ -30,7 +36,7 @@ class Observable<T> implements IObservab
         }
     }
 
-    wait(ct: ICancellation = Cancellation.none): Promise<T> {
+    next(ct: ICancellation = Cancellation.none): Promise<T> {
         return new Promise<T>((resolve, reject) => {
             this._once.push(resolve);
             if (ct.isSupported()) {
@@ -42,24 +48,28 @@ class Observable<T> implements IObservab
         });
     }
 
-    onObserverException(e: any) {
-        trace.error("Unhandled exception in the observer: {0}", e);
+    destroy() {
+        if(this._cleanup)
+            this._cleanup.call(null);
     }
 
-    private _removeOnce(d: IObserver<T>) {
+    protected onObserverException(e: any) {
+    }
+
+    private _removeOnce(d: Handler<T>) {
         let i = this._once.indexOf(d);
         if (i >= 0)
             this._once.splice(i);
     }
 
-    private _removeObserver(d: IObserver<T>) {
+    private _removeObserver(d: Handler<T>) {
         let i = this._observers.indexOf(d);
         if (i >= 0)
             this._observers.splice(i);
     }
 
-    private _notify(evt: T) {
-        let guard = (observer: IObserver<T>) => {
+    protected _notify(evt: T) {
+        let guard = (observer: Handler<T>) => {
             try {
                 observer(evt);
             } catch (e) {
@@ -79,7 +89,6 @@ class Observable<T> implements IObservab
 }
 
 namespace Observable {
-    export const traceSource = trace;
 }
 
 export = Observable;
\ No newline at end of file
diff --git a/src/ts/interfaces.ts b/src/ts/interfaces.ts
--- a/src/ts/interfaces.ts
+++ b/src/ts/interfaces.ts
@@ -1,5 +1,3 @@
-import { watchFile } from "fs";
-
 export interface IDestroyable {
     destroy();
 }
@@ -72,12 +70,7 @@ export interface ICancellable {
     cancel(reason?: any): void;
 }
 
-export interface IObserver<T> {
-    (x:T): void;
-}
-
 export interface IObservable<T> {
-    on(observer: IObserver<T>): IDestroyable;
-
-    wait(ct?: ICancellation) : Promise<T>;
+    on(next: (x:T) => void, error?: (e:any) => void, complete?:() => void): IDestroyable;
+    next(ct?: ICancellation) : Promise<T>;
 }
\ No newline at end of file
diff --git a/src/ts/log/TraceEvent.ts b/src/ts/log/TraceEvent.ts
new file mode 100644
--- /dev/null
+++ b/src/ts/log/TraceEvent.ts
@@ -0,0 +1,21 @@
+import * as TraceSource from './TraceSource'
+
+class TraceEvent {
+    readonly source: TraceSource;
+
+    readonly level: Number;
+
+    readonly arg: any;
+
+    constructor(source: TraceSource, level: Number, arg: any) {
+        this.source = source;
+        this.level = level;
+        this.arg = arg;
+    }
+}
+
+namespace TraceEvent {
+    
+}
+
+export = TraceEvent
\ No newline at end of file
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
@@ -1,17 +1,8 @@
 import * as format from '../text/format'
 import { argumentNotNull } from '../safe';
-
-interface TraceEventHandler {
-    (sender: TraceSource, level: number, arg: any): void;
-}
-
-interface TraceSourceHandler {
-    (source: TraceSource): void;
-}
-
-interface Destroyable {
-    destroy();
-}
+import * as Observable from '../components/Observable'
+import { IDestroyable } from '../interfaces';
+import * as TraceEvent from './TraceEvent'
 
 class Registry {
     static readonly instance = new Registry();
@@ -46,7 +37,7 @@ class Registry {
             this._listeners[i].call(null, source);
     }
 
-    on(handler: TraceSourceHandler): Destroyable {
+    on(handler: (source: TraceSource) => void): IDestroyable {
         argumentNotNull(handler, "handler");
         var me = this;
 
@@ -65,45 +56,18 @@ class Registry {
     }
 }
 
-class TraceSource {
-
+class TraceSource extends Observable<TraceEvent> {
     readonly id: any
 
-    // using array will provide faster iteration the with object
-    private _handlers: Array<TraceEventHandler> = new Array<TraceEventHandler>();
-
     level: number
 
     constructor(id: any) {
+        super();
         this.id = id || new Object();
     }
 
-    on(handler: TraceEventHandler): Destroyable {
-        argumentNotNull(handler, "handler");
-        var me = this;
-        me._handlers.push(handler);
-
-        return {
-            destroy() {
-                me.remove(handler);
-            }
-        }
-    }
-
-    remove(handler: TraceEventHandler): void {
-        let i = this._handlers.indexOf(handler);
-        if (i >= 0)
-            this._handlers.splice(i, 1);
-    }
-
     protected emit(level: number, arg: any) {
-        this._handlers.forEach(h => {
-            try {
-                h(this, level, arg);
-            } catch (e) {
-                // suppress error in log handlers
-            }
-        });
+        this._notify(new TraceEvent(this, level, arg));
     }
 
     isDebugEnabled() {
@@ -178,7 +142,7 @@ class TraceSource {
      * 
      * @param handler the handler which will be called for each trace source
      */
-    static on(handler: TraceSourceHandler) {
+    static on(handler: (source: TraceSource) => void) {
         return Registry.instance.on(handler);
     }
 
diff --git a/src/ts/log/writers/ConsoleWriter.ts b/src/ts/log/writers/ConsoleWriter.ts
new file mode 100644
--- /dev/null
+++ b/src/ts/log/writers/ConsoleWriter.ts
@@ -0,0 +1,29 @@
+import { IObservable } 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);
+        }
+    }
+
+    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);
+        }
+    }
+}
+
+namespace ConsoleWriter {
+}
+
+export = ConsoleWriter;
\ No newline at end of file
diff --git a/test/ts/ActivatableTests.ts b/test/ts/ActivatableTests.ts
--- a/test/ts/ActivatableTests.ts
+++ b/test/ts/ActivatableTests.ts
@@ -1,10 +1,8 @@
 import * as tape from 'tape';
 import * as ActivatableMixin from '@implab/core/components/ActivatableMixin';
 import { AsyncComponent } from '@implab/core/components/AsyncComponent';
-import { IActivationController } from '@implab/core/components/IActivationController';
-import { IActivatable } from '@implab/core/components/IActivatable';
-import { ICancellation } from '@implab/core/ICancellation';
-import { EmptyCancellation } from '@implab/core/EmptyCancellation';
+import { IActivationController, IActivatable, ICancellation } from '@implab/core/interfaces';
+import { Cancellation } from '@implab/core/Cancellation';
 
 class SimpleActivatable extends ActivatableMixin(AsyncComponent) {
 
@@ -33,20 +31,20 @@ class MockActivationController implement
         await component.activate();
     }
 
-    async activating(component: IActivatable, ct: ICancellation = EmptyCancellation.default) {
+    async activating(component: IActivatable, ct: ICancellation = Cancellation.none) {
         if (component != this._active)
             await this.deactivate();
     }
 
-    async activated(component: IActivatable, ct: ICancellation = EmptyCancellation.default) {
+    async activated(component: IActivatable, ct: ICancellation = Cancellation.none) {
         this._active = component;
     }
 
-    async deactivating(component: IActivatable, ct: ICancellation = EmptyCancellation.default) {
+    async deactivating(component: IActivatable, ct: ICancellation = Cancellation.none) {
 
     }
 
-    async deactivated(component: IActivatable, ct: ICancellation = EmptyCancellation.default) {
+    async deactivated(component: IActivatable, ct: ICancellation = Cancellation.none) {
         if (this._active == component)
             this._active = null;
     }
diff --git a/test/ts/TraceSourceTests.ts b/test/ts/TraceSourceTests.ts
--- a/test/ts/TraceSourceTests.ts
+++ b/test/ts/TraceSourceTests.ts
@@ -1,5 +1,6 @@
 import * as TraceSource from '@implab/core/log/TraceSource'
 import * as tape from 'tape';
+import * as ConsoleWriter from '@implab/core/log/writers/ConsoleWriter';
 
 const sourceId = 'test/TraceSourceTests';
 
@@ -8,10 +9,10 @@ tape('trace message', t => {
 
     trace.level = TraceSource.DebugLevel;
 
-    let h = trace.on((sender,level,msg) => {
-        t.equal(sender, trace, "sender should be the current trace source");
-        t.equal(TraceSource.DebugLevel, level, "level should be debug level");
-        t.equal(msg, "Hello, World!", "The message should be a formatted message");
+    let h = trace.on((ev) => {
+        t.equal(ev.source, trace, "sender should be the current trace source");
+        t.equal(ev.level, TraceSource.DebugLevel, "level should be debug level");
+        t.equal(ev.arg, "Hello, World!", "The message should be a formatted message");
 
         t.end();
     });
@@ -30,10 +31,10 @@ tape('trace event', t => {
         name: "custom event"
     };
 
-    let h = trace.on((sender,level,msg) => {
-        t.equal(sender, trace, "sender should be the current trace source");
-        t.equal(TraceSource.DebugLevel, level, "level should be debug level");
-        t.equal(msg, event, "The message should be the specified object");
+    let h = trace.on((ev) => {
+        t.equal(ev.source, trace, "sender should be the current trace source");
+        t.equal(ev.level, TraceSource.DebugLevel, "level should be debug level");
+        t.equal(ev.arg, event, "The message should be the specified object");
 
         t.end();
     });
@@ -41,4 +42,21 @@ tape('trace event', t => {
     trace.traceEvent(TraceSource.DebugLevel, event);
 
     h.destroy();
+});
+
+tape('console writer', async t => {
+    let writer = new ConsoleWriter();
+
+    let trace = TraceSource.get(sourceId);
+    trace.level = TraceSource.DebugLevel;
+
+    let p = writer.write(trace);
+
+    trace.log("Hello, {0}!", 'World');
+    trace.warn("Look at me!");
+    trace.error("DIE!");
+
+    console.log("DONE");
+
+    t.end();
 });
\ No newline at end of file