##// 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 import * as format from '../text/format'
1 import * as format from '../text/format'
2 import { argumentNotNull } from '../safe';
2 import { argumentNotNull } from '../safe';
3 import { Observable } from '../components/Observable'
3 import { Observable } from '../Observable'
4 import { IDestroyable } from '../interfaces';
4 import { IDestroyable } from '../interfaces';
5
5
6 export const DebugLevel = 400;
6 export const DebugLevel = 400;
7
7
8 export const LogLevel = 300;
8 export const LogLevel = 300;
9
9
10 export const WarnLevel = 200;
10 export const WarnLevel = 200;
11
11
12 export const ErrorLevel = 100;
12 export const ErrorLevel = 100;
13
13
14 export const SilentLevel = 0;
14 export const SilentLevel = 0;
15
15
16 export class TraceEvent {
16 export class TraceEvent {
17 readonly source: TraceSource;
17 readonly source: TraceSource;
18
18
19 readonly level: Number;
19 readonly level: Number;
20
20
21 readonly arg: any;
21 readonly arg: any;
22
22
23 constructor(source: TraceSource, level: Number, arg: any) {
23 constructor(source: TraceSource, level: Number, arg: any) {
24 this.source = source;
24 this.source = source;
25 this.level = level;
25 this.level = level;
26 this.arg = arg;
26 this.arg = arg;
27 }
27 }
28 }
28 }
29
29
30 class Registry {
30 class Registry {
31 static readonly instance = new Registry();
31 static readonly instance = new Registry();
32
32
33 private _registry: object = new Object();
33 private _registry: object = new Object();
34 private _listeners: object = new Object();
34 private _listeners: object = new Object();
35 private _nextCookie: number = 1;
35 private _nextCookie: number = 1;
36
36
37 get(id: any): TraceSource {
37 get(id: any): TraceSource {
38 argumentNotNull(id, "id");
38 argumentNotNull(id, "id");
39
39
40 if (this._registry[id])
40 if (this._registry[id])
41 return this._registry[id];
41 return this._registry[id];
42
42
43 var source = new TraceSource(id);
43 var source = new TraceSource(id);
44 this._registry[id] = source;
44 this._registry[id] = source;
45 this._onNewSource(source);
45 this._onNewSource(source);
46
46
47 return source;
47 return source;
48 }
48 }
49
49
50 add(id: any, source: TraceSource) {
50 add(id: any, source: TraceSource) {
51 argumentNotNull(id, "id");
51 argumentNotNull(id, "id");
52 argumentNotNull(source, "source");
52 argumentNotNull(source, "source");
53
53
54 this._registry[id] = source;
54 this._registry[id] = source;
55 this._onNewSource(source);
55 this._onNewSource(source);
56 }
56 }
57
57
58 _onNewSource(source: TraceSource) {
58 _onNewSource(source: TraceSource) {
59 for (let i in this._listeners)
59 for (let i in this._listeners)
60 this._listeners[i].call(null, source);
60 this._listeners[i].call(null, source);
61 }
61 }
62
62
63 on(handler: (source: TraceSource) => void): IDestroyable {
63 on(handler: (source: TraceSource) => void): IDestroyable {
64 argumentNotNull(handler, "handler");
64 argumentNotNull(handler, "handler");
65 var me = this;
65 var me = this;
66
66
67 var cookie = this._nextCookie++;
67 var cookie = this._nextCookie++;
68
68
69 this._listeners[cookie] = handler;
69 this._listeners[cookie] = handler;
70
70
71 for (let i in this._registry)
71 for (let i in this._registry)
72 handler(this._registry[i]);
72 handler(this._registry[i]);
73
73
74 return {
74 return {
75 destroy() {
75 destroy() {
76 delete me._listeners[cookie];
76 delete me._listeners[cookie];
77 }
77 }
78 };
78 };
79 }
79 }
80 }
80 }
81
81
82 export class TraceSource {
82 export class TraceSource {
83 readonly id: any
83 readonly id: any
84
84
85 level: number
85 level: number
86
86
87 readonly events: Observable<TraceEvent>
87 readonly events: Observable<TraceEvent>
88
88
89 _notifyNext: (arg: TraceEvent) => void
89 _notifyNext: (arg: TraceEvent) => void
90
90
91 constructor(id: any) {
91 constructor(id: any) {
92
92
93 this.id = id || new Object();
93 this.id = id || new Object();
94 this.events = new Observable((next) => {
94 this.events = new Observable((next) => {
95 this._notifyNext = next;
95 this._notifyNext = next;
96 })
96 })
97 }
97 }
98
98
99 protected emit(level: number, arg: any) {
99 protected emit(level: number, arg: any) {
100 this._notifyNext(new TraceEvent(this, level, arg));
100 this._notifyNext(new TraceEvent(this, level, arg));
101 }
101 }
102
102
103 isDebugEnabled() {
103 isDebugEnabled() {
104 return this.level >= DebugLevel;
104 return this.level >= DebugLevel;
105 }
105 }
106
106
107 debug(msg: string, ...args: any[]) {
107 debug(msg: string, ...args: any[]) {
108 if (this.isEnabled(DebugLevel))
108 if (this.isEnabled(DebugLevel))
109 this.emit(DebugLevel, format(msg, args));
109 this.emit(DebugLevel, format(msg, args));
110 }
110 }
111
111
112 isLogEnabled() {
112 isLogEnabled() {
113 return this.level >= LogLevel;
113 return this.level >= LogLevel;
114 }
114 }
115
115
116 log(msg: string, ...args: any[]) {
116 log(msg: string, ...args: any[]) {
117 if (this.isEnabled(LogLevel))
117 if (this.isEnabled(LogLevel))
118 this.emit(LogLevel, format(msg, args));
118 this.emit(LogLevel, format(msg, args));
119 }
119 }
120
120
121 isWarnEnabled() {
121 isWarnEnabled() {
122 return this.level >= WarnLevel;
122 return this.level >= WarnLevel;
123 }
123 }
124
124
125 warn(msg: string, ...args: any[]) {
125 warn(msg: string, ...args: any[]) {
126 if (this.isEnabled(WarnLevel))
126 if (this.isEnabled(WarnLevel))
127 this.emit(WarnLevel, format(msg, args));
127 this.emit(WarnLevel, format(msg, args));
128 }
128 }
129
129
130 /**
130 /**
131 * returns true if errors will be recorded.
131 * returns true if errors will be recorded.
132 */
132 */
133 isErrorEnabled() {
133 isErrorEnabled() {
134 return this.level >= ErrorLevel;
134 return this.level >= ErrorLevel;
135 }
135 }
136
136
137 /**
137 /**
138 * Traces a error.
138 * Traces a error.
139 *
139 *
140 * @param msg the message.
140 * @param msg the message.
141 * @param args parameters which will be substituted in the message.
141 * @param args parameters which will be substituted in the message.
142 */
142 */
143 error(msg: string, ...args: any[]) {
143 error(msg: string, ...args: any[]) {
144 if (this.isEnabled(ErrorLevel))
144 if (this.isEnabled(ErrorLevel))
145 this.emit(ErrorLevel, format(msg, args));
145 this.emit(ErrorLevel, format(msg, args));
146 }
146 }
147
147
148 /**
148 /**
149 * Checks whether the specified level is enabled for this
149 * Checks whether the specified level is enabled for this
150 * trace source.
150 * trace source.
151 *
151 *
152 * @param level the trace level which should be checked.
152 * @param level the trace level which should be checked.
153 */
153 */
154 isEnabled(level: number) {
154 isEnabled(level: number) {
155 return (this.level >= level);
155 return (this.level >= level);
156 }
156 }
157
157
158 /**
158 /**
159 * Traces a raw event, passing data as it is to the underlying listeners
159 * Traces a raw event, passing data as it is to the underlying listeners
160 *
160 *
161 * @param level the level of the event
161 * @param level the level of the event
162 * @param arg the data of the event, can be a simple string or any object.
162 * @param arg the data of the event, can be a simple string or any object.
163 */
163 */
164 traceEvent(level: number, arg: any) {
164 traceEvent(level: number, arg: any) {
165 if (this.isEnabled(level))
165 if (this.isEnabled(level))
166 this.emit(level, arg);
166 this.emit(level, arg);
167 }
167 }
168
168
169 /**
169 /**
170 * Register the specified handler to be called for every new and already
170 * Register the specified handler to be called for every new and already
171 * created trace source.
171 * created trace source.
172 *
172 *
173 * @param handler the handler which will be called for each trace source
173 * @param handler the handler which will be called for each trace source
174 */
174 */
175 static on(handler: (source: TraceSource) => void) {
175 static on(handler: (source: TraceSource) => void) {
176 return Registry.instance.on(handler);
176 return Registry.instance.on(handler);
177 }
177 }
178
178
179 /**
179 /**
180 * Creates or returns already created trace source for the specified id.
180 * Creates or returns already created trace source for the specified id.
181 *
181 *
182 * @param id the id for the trace source
182 * @param id the id for the trace source
183 */
183 */
184 static get(id: any) {
184 static get(id: any) {
185 return Registry.instance.get(id);
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
NO CONTENT: file was removed
Show file before | Show file after | Hide comments
1 NO CONTENT: file was removed
NO CONTENT: file was removed
General Comments 0
You need to be logged in to leave comments. Login now