##// 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,6 +1,6
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;
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