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

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

the documentation on observables is added...
cin -
r26:0b0a30e050ba propose observables
parent child
Show More
Show file before | Show file after | Hide comments
@@ -1,85 +1,135
1 1 # Observable
2 2
3 3 Универсальный способ организации потока сообщений. Данный механизм может
4 4 использоваться для оповещения об изменениях состояний объектов или для доставки
5 5 самостоятельных событий, например, связанных с действиями пользователя.
6 6
7 7 Является реализацией классического шаблона наблюдателя с возможность сообщить
8 8 о коце потока событий. Данная реализация не содержит никаких дополнительных
9 9 функций, таких как фильтрация, канал с состоянием, преобразования сообщений и
10 10 т.п. Это сделано специально, чтобы реализация оставалась максимально простой.
11 11
12 12 Пример того, как можно создать последовательность из 10 событий:
13 13
14 14 ```ts
15 15 var events = new Observable(async (notify, error, complete) => {
16 16 // цикл в котором возникает событие
17 17 for(let i = 0; i < 10; i++) {
18 18 await delay(1000);
19 19 // в качестве данных передается номер события
20 20 notify(i);
21 21 }
22 22 // по окончании последовательности информируем, что событий больше не будет
23 23 compelte();
24 24 });
25 25
26 26 // создаем окно с отображением хода событий
27 27 var progress = showProgress({ min: 0, max: 9, current: 0});
28 28
29 29 // подписываемся на события
30 30 events.on(
31 31 // обработчик очередного события
32 32 msg => {
33 33 progress.setValue(msg);
34 },
34 }.
35 35 // обработчик ошибки
36 36 e => {
37 37 progress.showError(e);
38 38 },
39 39 // обработчик конца потока
40 40 () => {
41 41 progress.close();
42 42 }
43 43 );
44
45 // ожидание следующего события
46 let firstEvent = await events.next();
44 47 ```
45 48
46 Пример создания `Observable` из событий другого объекта, например, виджета:
49 `Observable` можно создавать из событий другого объекта, например, виджета:
47 50
48 51 ```ts
52 // клсс
53 class Canvas {
54 readonly mouseMove: IObservable<[number,number]>
55
49 56 postCreate() {
50 57 // превращаем события виджета в Observable
51 this.mouseMove = new Observable((notify) => {
52 this.moveArea.on('mousemove',(x) => notify(x.) );
58 this.mouseMove = new Observable<[number,number]>((notify) => {
59 this.mousePad.on('mousemove',(e) => notify([e.clientX, e.clientY]) );
53 60 });
54 61 }
62 }
55 63
56 64 ```
57 65
58 Пример инициализации `Observable` внутри класса и генерация событий:
66 Если объект инкапсулирует в себе `Observable`, он также может сохранить методы
67 для оповещения подписчиков для дальнейшего их использования внутри класса.
59 68
60 69 ```ts
61
62 class PositionWidget extends Widget {
70 // класс, который будет генерировать события местоположения
71 class PositionTracker implements IDestroyable {
72 // _nextPosition и _complete будут связаны с position при создании
73 // экземпляра PositionTracker.
63 74 _nextPosition: (pos: Position) => void
64
65 75 _complete: () => void
66 76
67 readonly position: Observable<Position>;
77 readonly position: IObservable<Position>
68 78
69 constructor(...args[]) {
79 // конструктор
80 constructor(...args: any[]) {
70 81 super(args);
71 82
83 // создаем Observable
72 84 this.position = new Observable<Position>((notify, error, complete) => {
85 // сохраняем методы для оповещения о новом местоположении
73 86 this._nextPosition = notify;
87 // метод об оповещении конца потока событий
74 88 this._complete = complete
75 89 });
76 90 }
77 91
92 // метод для очистки ресурсов
78 93 destroy() {
79 94 this._complete();
80 95
81 96 super();
82 97 }
83 98 }
84 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
85 135 ``` No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,85 +1,186
1 1 # Observable
2 2
3 3 Универсальный способ организации потока сообщений. Данный механизм может
4 4 использоваться для оповещения об изменениях состояний объектов или для доставки
5 5 самостоятельных событий, например, связанных с действиями пользователя.
6 6
7 7 Является реализацией классического шаблона наблюдателя с возможность сообщить
8 8 о коце потока событий. Данная реализация не содержит никаких дополнительных
9 9 функций, таких как фильтрация, канал с состоянием, преобразования сообщений и
10 10 т.п. Это сделано специально, чтобы реализация оставалась максимально простой.
11 11
12 12 Пример того, как можно создать последовательность из 10 событий:
13 13
14 14 ```ts
15 15 var events = new Observable(async (notify, error, complete) => {
16 16 // цикл в котором возникает событие
17 17 for(let i = 0; i < 10; i++) {
18 18 await delay(1000);
19 19 // в качестве данных передается номер события
20 20 notify(i);
21 21 }
22 22 // по окончании последовательности информируем, что событий больше не будет
23 23 compelte();
24 24 });
25 25
26 26 // создаем окно с отображением хода событий
27 27 var progress = showProgress({ min: 0, max: 9, current: 0});
28 28
29 29 // подписываемся на события
30 30 events.on(
31 31 // обработчик очередного события
32 32 msg => {
33 33 progress.setValue(msg);
34 },
34 }.
35 35 // обработчик ошибки
36 36 e => {
37 37 progress.showError(e);
38 38 },
39 39 // обработчик конца потока
40 40 () => {
41 41 progress.close();
42 42 }
43 43 );
44
45 // ожидание следующего события
46 let firstEvent = await events.next();
44 47 ```
45 48
46 Пример создания `Observable` из событий другого объекта, например, виджета:
49 `Observable` можно создавать из событий другого объекта, например, виджета:
47 50
48 51 ```ts
52 // клсс
53 class Canvas {
54 readonly mouseMove: IObservable<[number,number]>
55
49 56 postCreate() {
50 57 // превращаем события виджета в Observable
51 this.mouseMove = new Observable((notify) => {
52 this.moveArea.on('mousemove',(x) => notify(x.) );
58 this.mouseMove = new Observable<[number,number]>((notify) => {
59 this.mousePad.on('mousemove',(e) => notify([e.clientX, e.clientY]) );
53 60 });
54 61 }
62 }
55 63
56 64 ```
57 65
58 Пример инициализации `Observable` внутри класса и генерация событий:
66 Если объект инкапсулирует в себе `Observable`, он также может сохранить методы
67 для оповещения подписчиков для дальнейшего их использования внутри класса.
59 68
60 69 ```ts
61
62 class PositionWidget extends Widget {
70 // класс, который будет генерировать события местоположения
71 class PositionTracker implements IDestroyable {
72 // _nextPosition и _complete будут связаны с position при создании
73 // экземпляра PositionTracker.
63 74 _nextPosition: (pos: Position) => void
64
65 75 _complete: () => void
66 76
67 readonly position: Observable<Position>;
77 readonly position: IObservable<Position>
68 78
69 constructor(...args[]) {
79 // конструктор
80 constructor(...args: any[]) {
70 81 super(args);
71 82
83 // создаем Observable
72 84 this.position = new Observable<Position>((notify, error, complete) => {
85 // сохраняем методы для оповещения о новом местоположении
73 86 this._nextPosition = notify;
87 // метод об оповещении конца потока событий
74 88 this._complete = complete
75 89 });
76 90 }
77 91
92 // метод для очистки ресурсов
78 93 destroy() {
79 94 this._complete();
80 95
81 96 super();
82 97 }
83 98 }
99 ```
84 100
101 Существует также несколько варинатов получения сообщений
102
103 ```ts
104 // регистрация метода для получений событий
105 let subscription = pushEvents.on((msg) => {
106 displayPopup(msg);
107 });
108
109 // подписку можно отменить, после чего обработчики больше не будут вызываться
110 subcription.destroy();
111
112 // если требуется получить только одно сообщение можно использовать
113 // асинхронный метод next(ct?: ICancellation)
114
115 let msg = await pushEvents.next();
116
117 // пример метода для получения координат с карты, который использует
118 // событие нажатия мышью для определения координат.
119
120 class Map {
121 /**
122
123 Получает координаты по щелчку мыши.
124
125 @async
126
127 @returns [lon,lat]
128
129 */
130 async peekCoordinates(ct: ICancellation = Cancellation.none) {
131 // получаем событие клика
132 let evt = this.viewport.click.next(ct);
133
134 // преобразуем позицию на экране в координаты карты
135 return this.clientToCoodinates([evt.clientx,evt.clientY]);
136 }
137 }
138
139
140 let map : Map; // где-то объявлено
141
142 // пример получения координат с карты
143 let coords = await map.peekCoordinates();
144
145 ```
146
147 ## Observable и последовательности
148
149 Можно сичтать, что `Observable` это некоторая аналогия итератора только в
150 парадигме событийного (или реактивного) программировния. Следует также понимать,
151 что при переходе от синхронного процедурного программирования к событийному так
152 же меняется и направление управления (Inverse Of Control), что означает
153 следующее:
154
155 * при работе с итераторами клиенты сами определяют момент чтения следующего
156 элемента последовательности.
157 * при работе с `Observable` клиенты вынуждены обрабатывать эти события по мере
158 их поступления и не могут на это повлиять.
159
160 Последний пункт можно изменить применив, например, буффер или канал с
161 состоянием, т.е. очередь, но данные механизмы выходят за рамки простого шаблона
162 наблюдателя.
163
164 ```ts
165 // обработка в цикле не гарантирует получения всех сообщений
166 while(1) {
167 // ожидаем следующее событие, по сути это подписка только на одно событие
168 let next = await events.next();
169
170 // такой цикл может пропускать сообщения, поскольку асинхронная операция
171 // позволит возобновить создание новых событий, на которые мы не подписаны
172 await processEvent(next);
173
174 // не только асинхронные операции могут привести к пропуску события
175 // например вызов метода, который приводит к созданию события так же
176 // приведет к тому, что созданное событие не будет обработано в текущем
177 // цикле
178 doSmthAndRiseEvent();
179 }
180
181 // для получения всех сообщений нужно регистрировать подписчика
182 events.on((data) => {
183 // будет вызван для всех сообщений
184 processEvent(data);
185 });
85 186 ``` No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,193 +1,195
1 1 import { IObservable, IDestroyable, ICancellation } from './interfaces';
2 2 import { Cancellation } from './Cancellation'
3 3 import { argumentNotNull } from './safe';
4 4
5 5
6 6 interface Handler<T> {
7 7 (x: T): void
8 8 }
9 9
10 10 interface Initializer<T> {
11 11 (notify: Handler<T>, error?: (e: any) => void, complete?: () => void): void;
12 12 }
13 13
14 14 // TODO: think about to move this interfaces.ts and make it public
15 15 interface IObserver<T> {
16 16 next(event: T): void
17 17
18 18 error(e: any): void
19 19
20 20 complete(): void
21 21 }
22 22
23 23 const noop = () => {};
24 24
25 25 export class Observable<T> implements IObservable<T> {
26 26 private _once = new Array<IObserver<T>>();
27 27
28 28 private _observers = new Array<IObserver<T>>();
29 29
30 30
31 31 private _complete: boolean
32 32
33 33 private _error: any
34 34
35 35 constructor(func?: Initializer<T>) {
36 36 if (func)
37 37 func(
38 38 this._notifyNext.bind(this),
39 39 this._notifyError.bind(this),
40 40 this._notifyCompleted.bind(this)
41 41 );
42 42 }
43 43
44 44 /**
45 45 * Registers handlers for the current observable object.
46 46 *
47 47 * @param next the handler for events
48 48 * @param error the handler for a error
49 49 * @param complete the handler for a completion
50 50 * @returns {IDestroyable} the handler for the current subscription, this
51 51 * handler can be used to unsubscribe from events.
52 52 *
53 53 */
54 54 on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
55 55 argumentNotNull(next, "next");
56 56
57 57 let me = this;
58 58
59 59 let observer: IObserver<T> & IDestroyable = {
60 60 next: next,
61 61 error: error ? error.bind(null) : noop,
62 62 complete: complete ? complete.bind(null) : noop,
63 63
64 64 destroy() {
65 65 me._removeObserver(this);
66 66 }
67 67 };
68 68
69 69 this._addObserver(observer);
70 70
71 71
72 72 return observer;
73 73 }
74 74
75 75 private _addObserver(observer: IObserver<T>) {
76 76 if (this._complete) {
77 77 try {
78 78 if (this._error)
79 79 observer.error(this._error);
80 80 else
81 81 observer.complete();
82 82 } catch (e) {
83 83 this.onObserverException(e);
84 84 }
85 85 } else {
86 86 this._observers.push(observer);
87 87 }
88 88 }
89 89
90 90 /**
91 91 * Waits for the next event. This method can't be used to read messages
92 92 * as a sequence since it can skip some messages between calls.
93 93 *
94 94 * @param ct a cancellation token
95 95 */
96 96 next(ct: ICancellation = Cancellation.none): Promise<T> {
97 97 return new Promise<T>((resolve, reject) => {
98 98 let observer: IObserver<T> = {
99 99 next: resolve,
100 100 error: reject,
101 101 complete: () => reject("No more events are available")
102 102 };
103 103
104 104 if (this._addOnce(observer) && ct.isSupported()) {
105 105 ct.register((e) => {
106 106 this._removeOnce(observer);
107 107 reject(e);
108 108 });
109 109 }
110 110 });
111 111 }
112 112
113 113 private _addOnce(observer: IObserver<T>) {
114 114 if (this._complete) {
115 115 try {
116 116 if (this._error)
117 117 observer.error(this._error);
118 118 else
119 119 observer.complete();
120 120 } catch (e) {
121 121 this.onObserverException(e);
122 122 }
123 123 return false;
124 124 }
125 125
126 126 this._once.push(observer);
127 127 return true;
128 128 }
129 129
130 130 protected onObserverException(e: any) {
131 131 }
132 132
133 133 private _removeOnce(d: IObserver<T>) {
134 134 let i = this._once.indexOf(d);
135 135 if (i >= 0)
136 136 this._once.splice(i, 1);
137 137 }
138 138
139 139 private _removeObserver(d: IObserver<T>) {
140 140 let i = this._observers.indexOf(d);
141 141 if (i >= 0)
142 142 this._observers.splice(i, 1);
143 143 }
144 144
145 145 private _notify(guard: (observer: IObserver<T>) => void) {
146 146 if (this._once.length) {
147 147 for (let i = 0; i < this._once.length; i++)
148 148 guard(this._once[i]);
149 149 this._once = [];
150 150 }
151 151
152 152 for (let i = 0; i < this._observers.length; i++)
153 153 guard(this._observers[i]);
154 154 }
155 155
156 156 protected _notifyNext(evt: T) {
157 157 let guard = (observer: IObserver<T>) => {
158 158 try {
159 159 observer.next(evt);
160 160 } catch (e) {
161 161 this.onObserverException(e);
162 162 }
163 163 }
164 164
165 165 this._notify(guard);
166 166 }
167 167
168 168 protected _notifyError(e: any) {
169 169 let guard = (observer: IObserver<T>) => {
170 170 try {
171 171 observer.error(e);
172 172 } catch (e) {
173 173 this.onObserverException(e);
174 174 }
175 175 }
176 176
177 177 this._notify(guard);
178 178 this._observers = [];
179 this._complete = true;
179 180 }
180 181
181 182 protected _notifyCompleted() {
182 183 let guard = (observer: IObserver<T>) => {
183 184 try {
184 185 observer.complete();
185 186 } catch (e) {
186 187 this.onObserverException(e);
187 188 }
188 189 }
189 190
190 191 this._notify(guard);
191 192 this._observers = [];
193 this._complete = true;
192 194 }
193 195 } No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,36 +1,74
1 1 import { TraceSource, DebugLevel } from '@implab/core/log/TraceSource'
2 2 import * as tape from 'tape';
3 3 import { TapeWriter, delay } from './TestTraits';
4 4 import { Observable } from '@implab/core/Observable';
5 5 import { IObservable } from '@implab/core/interfaces';
6 6
7 7 let trace = TraceSource.get("ObservableTests");
8 8
9 9 tape('events sequence example', async t => {
10 10
11 11
12 12 let events: IObservable<number>
13 13
14 14 let done = new Promise<void>((resolve) => {
15 15 events = new Observable<number>(async (notify, fail, complete) => {
16 16 for (let i = 0; i < 10; i++) {
17 17 await delay(0);
18 18 notify(i);
19 19 }
20 complete();
20 21 resolve();
21 22 });
22 23 });
23 24
24 25 let count = 0;
25 events.on(x => count = count + x);
26 let complete = false;
27 events.on(x => count = count + x, null, () => complete = true);
26 28
27 29 let first = await events.next();
28 30
29 31 t.equals(first, 0, "the first event");
32 t.false(complete, "the sequence is not complete");
30 33
31 34 await done;
32 35
33 36 t.equals(count, 45, "the summ of the evetns");
37 t.true(complete, "the sequence is complete");
38
39 t.end();
40 });
41
42 tape('event sequence termination', async t => {
43 let events: IObservable<number>
44
45 let done = new Promise<void>((resolve) => {
46 events = new Observable<number>(async (notify, fail, complete) => {
47 await delay(0);
48 notify(1);
49 complete();
50 notify(2);
51 complete();
52 fail("Sequence terminated");
53 resolve();
54 });
55 });
56
57 let count = 0;
58 events.on(() => {}, (e) => count++, () => count++);
59
60 let first = await events.next();
61 t.equals(first, 1, "the first message");
62 try {
63 await events.next();
64 t.fail("shoud throw an exception");
65 } catch(e) {
66 t.pass("the sequence is terminated");
67 }
68
69 await done;
70
71 t.equals(count, 1, "the sequence must be terminated once");
34 72
35 73 t.end();
36 74 }); No newline at end of file
General Comments 0
You need to be logged in to leave comments. Login now