##// END OF EJS Templates
implab » implabjs-core
RSS

Clone from http://hg.code.sf.net/p/implabjs/core

documenting...
cin -
r23:d420abe33095 propose observables
parent child
Show More
Show file before | Show file after | Hide comments
@@ -0,0 +1,135
1 # Observable
2
3 Универсальный способ организации потока сообщений. Данный механизм может
4 использоваться для оповещения об изменениях состояний объектов или для доставки
5 самостоятельных событий, например, связанных с действиями пользователя.
6
7 Является реализацией классического шаблона наблюдателя с возможность сообщить
8 о коце потока событий. Данная реализация не содержит никаких дополнительных
9 функций, таких как фильтрация, канал с состоянием, преобразования сообщений и
10 т.п. Это сделано специально, чтобы реализация оставалась максимально простой.
11
12 Пример того, как можно создать последовательность из 10 событий:
13
14 ```ts
15 var events = new Observable(async (notify, error, complete) => {
16 // цикл в котором возникает событие
17 for(let i = 0; i < 10; i++) {
18 await delay(1000);
19 // в качестве данных передается номер события
20 notify(i);
21 }
22 // по окончании последовательности информируем, что событий больше не будет
23 compelte();
24 });
25
26 // создаем окно с отображением хода событий
27 var progress = showProgress({ min: 0, max: 9, current: 0});
28
29 // подписываемся на события
30 events.on(
31 // обработчик очередного события
32 msg => {
33 progress.setValue(msg);
34 }.
35 // обработчик ошибки
36 e => {
37 progress.showError(e);
38 },
39 // обработчик конца потока
40 () => {
41 progress.close();
42 }
43 );
44
45 // ожидание следующего события
46 let firstEvent = await events.next();
47 ```
48
49 `Observable` можно создавать из событий другого объекта, например, виджета:
50
51 ```ts
52 // клсс
53 class Canvas {
54 readonly mouseMove: IObservable<[number,number]>
55
56 postCreate() {
57 // превращаем события виджета в Observable
58 this.mouseMove = new Observable<[number,number]>((notify) => {
59 this.mousePad.on('mousemove',(e) => notify([e.clientX, e.clientY]) );
60 });
61 }
62 }
63
64 ```
65
66 Если объект инкапсулирует в себе `Observable`, он также может сохранить методы
67 для оповещения подписчиков для дальнейшего их использования внутри класса.
68
69 ```ts
70 // класс, который будет генерировать события местоположения
71 class PositionTracker implements IDestroyable {
72 // _nextPosition и _complete будут связаны с position при создании
73 // экземпляра PositionTracker.
74 _nextPosition: (pos: Position) => void
75 _complete: () => void
76
77 readonly position: IObservable<Position>
78
79 // конструктор
80 constructor(...args: any[]) {
81 super(args);
82
83 // создаем Observable
84 this.position = new Observable<Position>((notify, error, complete) => {
85 // сохраняем методы для оповещения о новом местоположении
86 this._nextPosition = notify;
87 // метод об оповещении конца потока событий
88 this._complete = complete
89 });
90 }
91
92 // метод для очистки ресурсов
93 destroy() {
94 this._complete();
95
96 super();
97 }
98 }
99
100 ```
101
102 ## Observable и последовательности
103
104 Можно сичтать, что `Observable` это некоторая аналогия итератора только в
105 парадигме событийного (или реактивного) программировния. Следует также понимать,
106 что при переходе от синхронного процедурного программирования к событийному так
107 же меняется и направление управления (Inverse Of Control), что означает
108 следующее:
109
110 * при работе с итераторами клиенты сами определяют момент чтения следующего
111 элемента последовательности.
112 * при работе с `Observable` клиенты вынуждены обрабатывать эти события по мере
113 их поступления и не могут на это повлиять.
114
115 Последний пункт можно изменить применив, например, буффер или канал с
116 состоянием, т.е. очередь, но данные механизмы выходят за рамки простого шаблона
117 наблюдателя.
118
119 ```ts
120 while(1) {
121 // ожидаем следующее событие, по сути это подписка только на одно событие
122 let next = await events.next();
123
124 // такой цикл может пропускать сообщения, поскольку асинхронная операция
125 // позволит возобновить создание новых событий, на которые мы не подписаны
126 await processEvent(next);
127
128 // не только асинхронные операции могут привести к пропуску события
129 // например вызов метода, который приводит к созданию события так же
130 // приведет к тому, что созданное событие не будет обработано в текущем
131 // цикле
132 doSmthAndRiseEvent();
133 }
134
135 ``` No newline at end of file
Show file before | Show file after | Hide comments
@@ -0,0 +1,193
1 import { IObservable, IDestroyable, ICancellation } from './interfaces';
2 import { Cancellation } from './Cancellation'
3 import { argumentNotNull } from './safe';
4
5
6 interface Handler<T> {
7 (x: T): void
8 }
9
10 interface Initializer<T> {
11 (notify: Handler<T>, error?: (e: any) => void, complete?: () => void): void;
12 }
13
14 // TODO: think about to move this interfaces.ts and make it public
15 interface IObserver<T> {
16 next(event: T): void
17
18 error(e: any): void
19
20 complete(): void
21 }
22
23 const noop = () => {};
24
25 export class Observable<T> implements IObservable<T> {
26 private _once = new Array<IObserver<T>>();
27
28 private _observers = new Array<IObserver<T>>();
29
30
31 private _complete: boolean
32
33 private _error: any
34
35 constructor(func?: Initializer<T>) {
36 if (func)
37 func(
38 this._notifyNext.bind(this),
39 this._notifyError.bind(this),
40 this._notifyCompleted.bind(this)
41 );
42 }
43
44 /**
45 * Registers handlers for the current observable object.
46 *
47 * @param next the handler for events
48 * @param error the handler for a error
49 * @param complete the handler for a completion
50 * @returns {IDestroyable} the handler for the current subscription, this
51 * handler can be used to unsubscribe from events.
52 *
53 */
54 on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
55 argumentNotNull(next, "next");
56
57 let me = this;
58
59 let observer: IObserver<T> & IDestroyable = {
60 next: next,
61 error: error ? error.bind(null) : noop,
62 complete: complete ? complete.bind(null) : noop,
63
64 destroy() {
65 me._removeObserver(this);
66 }
67 }
68
69 this._addObserver(observer);
70
71
72 return observer;
73 }
74
75 private _addObserver(observer: IObserver<T>) {
76 if (this._complete) {
77 try {
78 if (this._error)
79 observer.error(this._error);
80 else
81 observer.complete();
82 } catch (e) {
83 this.onObserverException(e);
84 }
85 } else {
86 this._observers.push(observer);
87 }
88 }
89
90 /**
91 * Waits for the next event. This method can't be used to read messages
92 * as a sequence since it can skip some messages between calls.
93 *
94 * @param ct a cancellation token
95 */
96 next(ct: ICancellation = Cancellation.none): Promise<T> {
97 return new Promise<T>((resolve, reject) => {
98 let observer: IObserver<T> = {
99 next: resolve,
100 error: reject,
101 complete: () => reject("No more events are available")
102 };
103
104 if (this._addOnce(observer) && ct.isSupported()) {
105 ct.register((e) => {
106 this._removeOnce(observer);
107 reject(e);
108 });
109 }
110 });
111 }
112
113 private _addOnce(observer: IObserver<T>) {
114 if (this._complete) {
115 try {
116 if (this._error)
117 observer.error(this._error);
118 else
119 observer.complete();
120 } catch (e) {
121 this.onObserverException(e);
122 }
123 return false;
124 }
125
126 this._once.push(observer);
127 return true;
128 }
129
130 protected onObserverException(e: any) {
131 }
132
133 private _removeOnce(d: IObserver<T>) {
134 let i = this._once.indexOf(d);
135 if (i >= 0)
136 this._once.splice(i, 1);
137 }
138
139 private _removeObserver(d: IObserver<T>) {
140 let i = this._observers.indexOf(d);
141 if (i >= 0)
142 this._observers.splice(i, 1);
143 }
144
145 private _notify(guard: (observer: IObserver<T>) => void) {
146 if (this._once.length) {
147 for (let i = 0; i < this._once.length; i++)
148 guard(this._once[i]);
149 this._once = [];
150 }
151
152 for (let i = 0; i < this._observers.length; i++)
153 guard(this._observers[i]);
154 }
155
156 protected _notifyNext(evt: T) {
157 let guard = (observer: IObserver<T>) => {
158 try {
159 observer.next(evt);
160 } catch (e) {
161 this.onObserverException(e);
162 }
163 }
164
165 this._notify(guard);
166 }
167
168 protected _notifyError(e: any) {
169 let guard = (observer: IObserver<T>) => {
170 try {
171 observer.error(e);
172 } catch (e) {
173 this.onObserverException(e);
174 }
175 }
176
177 this._notify(guard);
178 this._observers = [];
179 }
180
181 protected _notifyCompleted() {
182 let guard = (observer: IObserver<T>) => {
183 try {
184 observer.complete();
185 } catch (e) {
186 this.onObserverException(e);
187 }
188 }
189
190 this._notify(guard);
191 this._observers = [];
192 }
193 } No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,188 +1,188
1 1 import * as format from '../text/format'
2 2 import { argumentNotNull } from '../safe';
3 import { Observable } from '../components/Observable'
3 import { Observable } from '../Observable'
4 4 import { IDestroyable } from '../interfaces';
5 5
6 6 export const DebugLevel = 400;
7 7
8 8 export const LogLevel = 300;
9 9
10 10 export const WarnLevel = 200;
11 11
12 12 export const ErrorLevel = 100;
13 13
14 14 export const SilentLevel = 0;
15 15
16 16 export class TraceEvent {
17 17 readonly source: TraceSource;
18 18
19 19 readonly level: Number;
20 20
21 21 readonly arg: any;
22 22
23 23 constructor(source: TraceSource, level: Number, arg: any) {
24 24 this.source = source;
25 25 this.level = level;
26 26 this.arg = arg;
27 27 }
28 28 }
29 29
30 30 class Registry {
31 31 static readonly instance = new Registry();
32 32
33 33 private _registry: object = new Object();
34 34 private _listeners: object = new Object();
35 35 private _nextCookie: number = 1;
36 36
37 37 get(id: any): TraceSource {
38 38 argumentNotNull(id, "id");
39 39
40 40 if (this._registry[id])
41 41 return this._registry[id];
42 42
43 43 var source = new TraceSource(id);
44 44 this._registry[id] = source;
45 45 this._onNewSource(source);
46 46
47 47 return source;
48 48 }
49 49
50 50 add(id: any, source: TraceSource) {
51 51 argumentNotNull(id, "id");
52 52 argumentNotNull(source, "source");
53 53
54 54 this._registry[id] = source;
55 55 this._onNewSource(source);
56 56 }
57 57
58 58 _onNewSource(source: TraceSource) {
59 59 for (let i in this._listeners)
60 60 this._listeners[i].call(null, source);
61 61 }
62 62
63 63 on(handler: (source: TraceSource) => void): IDestroyable {
64 64 argumentNotNull(handler, "handler");
65 65 var me = this;
66 66
67 67 var cookie = this._nextCookie++;
68 68
69 69 this._listeners[cookie] = handler;
70 70
71 71 for (let i in this._registry)
72 72 handler(this._registry[i]);
73 73
74 74 return {
75 75 destroy() {
76 76 delete me._listeners[cookie];
77 77 }
78 78 };
79 79 }
80 80 }
81 81
82 82 export class TraceSource {
83 83 readonly id: any
84 84
85 85 level: number
86 86
87 87 readonly events: Observable<TraceEvent>
88 88
89 89 _notifyNext: (arg: TraceEvent) => void
90 90
91 91 constructor(id: any) {
92 92
93 93 this.id = id || new Object();
94 94 this.events = new Observable((next) => {
95 95 this._notifyNext = next;
96 96 })
97 97 }
98 98
99 99 protected emit(level: number, arg: any) {
100 100 this._notifyNext(new TraceEvent(this, level, arg));
101 101 }
102 102
103 103 isDebugEnabled() {
104 104 return this.level >= DebugLevel;
105 105 }
106 106
107 107 debug(msg: string, ...args: any[]) {
108 108 if (this.isEnabled(DebugLevel))
109 109 this.emit(DebugLevel, format(msg, args));
110 110 }
111 111
112 112 isLogEnabled() {
113 113 return this.level >= LogLevel;
114 114 }
115 115
116 116 log(msg: string, ...args: any[]) {
117 117 if (this.isEnabled(LogLevel))
118 118 this.emit(LogLevel, format(msg, args));
119 119 }
120 120
121 121 isWarnEnabled() {
122 122 return this.level >= WarnLevel;
123 123 }
124 124
125 125 warn(msg: string, ...args: any[]) {
126 126 if (this.isEnabled(WarnLevel))
127 127 this.emit(WarnLevel, format(msg, args));
128 128 }
129 129
130 130 /**
131 131 * returns true if errors will be recorded.
132 132 */
133 133 isErrorEnabled() {
134 134 return this.level >= ErrorLevel;
135 135 }
136 136
137 137 /**
138 138 * Traces a error.
139 139 *
140 140 * @param msg the message.
141 141 * @param args parameters which will be substituted in the message.
142 142 */
143 143 error(msg: string, ...args: any[]) {
144 144 if (this.isEnabled(ErrorLevel))
145 145 this.emit(ErrorLevel, format(msg, args));
146 146 }
147 147
148 148 /**
149 149 * Checks whether the specified level is enabled for this
150 150 * trace source.
151 151 *
152 152 * @param level the trace level which should be checked.
153 153 */
154 154 isEnabled(level: number) {
155 155 return (this.level >= level);
156 156 }
157 157
158 158 /**
159 159 * Traces a raw event, passing data as it is to the underlying listeners
160 160 *
161 161 * @param level the level of the event
162 162 * @param arg the data of the event, can be a simple string or any object.
163 163 */
164 164 traceEvent(level: number, arg: any) {
165 165 if (this.isEnabled(level))
166 166 this.emit(level, arg);
167 167 }
168 168
169 169 /**
170 170 * Register the specified handler to be called for every new and already
171 171 * created trace source.
172 172 *
173 173 * @param handler the handler which will be called for each trace source
174 174 */
175 175 static on(handler: (source: TraceSource) => void) {
176 176 return Registry.instance.on(handler);
177 177 }
178 178
179 179 /**
180 180 * Creates or returns already created trace source for the specified id.
181 181 *
182 182 * @param id the id for the trace source
183 183 */
184 184 static get(id: any) {
185 185 return Registry.instance.get(id);
186 186 }
187 187 }
188 188
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
General Comments 0
You need to be logged in to leave comments. Login now