implab/implabjs-core Commit - r26:0b0a30e050ba

the documentation on observables is added...

cin -

r26:0b0a30e050ba propose observables

parent child

Context file:

r26:0b0a30e050ba

Collapse all files

docs/en/observable.md +60 -10

             # Observable
             Универсальный способ организации потока сообщений. Данный механизм может
             использоваться для оповещения об изменениях состояний объектов или для доставки
             самостоятельных событий, например, связанных с действиями пользователя.
             Является реализацией классического шаблона наблюдателя с возможность сообщить
             о коце потока событий. Данная реализация не содержит никаких дополнительных
             функций, таких как фильтрация, канал с состоянием, преобразования сообщений и
             т.п. Это сделано специально, чтобы реализация оставалась максимально простой.
             Пример того, как можно создать последовательность из 10 событий:
             ```ts
             var events = new Observable(async (notify, error, complete) => {
                 // цикл в котором возникает событие
                 for(let i = 0; i < 10; i++) {
                     await delay(1000);
                     // в качестве данных передается номер события
                     notify(i);
                 }
                 // по окончании последовательности информируем, что событий больше не будет
                 compelte();
             });
             // создаем окно с отображением хода событий
             var progress = showProgress({ min: 0, max: 9, current: 0});
             // подписываемся на события
             events.on(
                 // обработчик очередного события
                 msg => {
                     progress.setValue(msg);
-                },
+                }.
                 // обработчик ошибки
                 e => {
                     progress.showError(e);
                 },
                 // обработчик конца потока
                 () => {
                     progress.close();
                 }
             );
+            // ожидание следующего события
+            let firstEvent = await events.next();
             ```
-            Пример создания `Observable` из событий другого объекта, например, виджета:
+            `Observable` можно создавать из событий другого объекта, например, виджета:
             ```ts
+            // клсс
+            class Canvas {
+                readonly mouseMove: IObservable<[number,number]>
                 postCreate() {
                     // превращаем события виджета в Observable
-                this.mouseMove = new Observable((notify) => {
+                    this.mouseMove = new Observable<[number,number]>((notify) => {
-                    this.moveArea.on('mousemove',(x) => notify(x.) );
+                        this.mousePad.on('mousemove',(e) => notify([e.clientX, e.clientY]) );
                     });
                 }
+            }
             ```
-            Пример инициализации `Observable` внутри класса и генерация событий:
+            Если объект инкапсулирует в себе `Observable`, он также может сохранить методы
+            для оповещения подписчиков для дальнейшего их использования внутри класса.
             ```ts
+            // класс, который будет генерировать события местоположения
-            class PositionWidget extends Widget {
+            class PositionTracker implements IDestroyable {
+                // _nextPosition и _complete будут связаны с position при создании
+                // экземпляра PositionTracker.
                 _nextPosition: (pos: Position) => void
                 _complete: () => void
-                readonly position: Observable<Position>;
+                readonly position: IObservable<Position>
-                constructor(...args[]) {
+                // конструктор
+                constructor(...args: any[]) {
                     super(args);
+                    // создаем Observable
                     this.position = new Observable<Position>((notify, error, complete) => {
+                        // сохраняем методы для оповещения о новом местоположении
                         this._nextPosition = notify;
+                        // метод об оповещении конца потока событий
                         this._complete = complete
                     });
                 }
+                // метод для очистки ресурсов
                 destroy() {
                     this._complete();
                     super();
                 }
             }
+            ```
+            ## Observable и последовательности
+            Можно сичтать, что `Observable` это некоторая аналогия итератора только в
+            парадигме событийного (или реактивного) программировния. Следует также понимать,
+            что при переходе от синхронного процедурного программирования к событийному так
+            же меняется и направление управления (Inverse Of Control), что означает
+            следующее:
+            * при работе с итераторами клиенты сами определяют момент чтения следующего
+              элемента последовательности.
+            * при работе с `Observable` клиенты вынуждены обрабатывать эти события по мере
+              их поступления и не могут на это повлиять.
+            Последний пункт можно изменить применив, например, буффер или канал с
+            состоянием, т.е. очередь, но данные механизмы выходят за рамки простого шаблона
+            наблюдателя.
+            ```ts
+            while(1) {
+                // ожидаем следующее событие, по сути это подписка только на одно событие
+                let next = await events.next();
+                // такой цикл может пропускать сообщения, поскольку асинхронная операция
+                // позволит возобновить создание новых событий, на которые мы не подписаны
+                await processEvent(next);
+                // не только асинхронные операции могут привести к пропуску события
+                // например вызов метода, который приводит к созданию события так же
+                // приведет к тому, что созданное событие не будет обработано в текущем
+                // цикле
+                doSmthAndRiseEvent();
+            }
             ```

docs/ru/observable.md +111 -10

@@ -1,85 +1,186
1	# Observable	1	# Observable
2		2
3	Универсальный способ организации потока сообщений. Данный механизм может	3	Универсальный способ организации потока сообщений. Данный механизм может
4	использоваться для оповещения об изменениях состояний объектов или для доставки	4	использоваться для оповещения об изменениях состояний объектов или для доставки
5	самостоятельных событий, например, связанных с действиями пользователя.	5	самостоятельных событий, например, связанных с действиями пользователя.
6		6
7	Является реализацией классического шаблона наблюдателя с возможность сообщить	7	Является реализацией классического шаблона наблюдателя с возможность сообщить
8	о коце потока событий. Данная реализация не содержит никаких дополнительных	8	о коце потока событий. Данная реализация не содержит никаких дополнительных
9	функций, таких как фильтрация, канал с состоянием, преобразования сообщений и	9	функций, таких как фильтрация, канал с состоянием, преобразования сообщений и
10	т.п. Это сделано специально, чтобы реализация оставалась максимально простой.	10	т.п. Это сделано специально, чтобы реализация оставалась максимально простой.
11		11
12	Пример того, как можно создать последовательность из 10 событий:	12	Пример того, как можно создать последовательность из 10 событий:
13		13
14	```ts	14	```ts
15	var events = new Observable(async (notify, error, complete) => {	15	var events = new Observable(async (notify, error, complete) => {
16	// цикл в котором возникает событие	16	// цикл в котором возникает событие
17	for(let i = 0; i < 10; i++) {	17	for(let i = 0; i < 10; i++) {
18	await delay(1000);	18	await delay(1000);
19	// в качестве данных передается номер события	19	// в качестве данных передается номер события
20	notify(i);	20	notify(i);
21	}	21	}
22	// по окончании последовательности информируем, что событий больше не будет	22	// по окончании последовательности информируем, что событий больше не будет
23	compelte();	23	compelte();
24	});	24	});
25		25
26	// создаем окно с отображением хода событий	26	// создаем окно с отображением хода событий
27	var progress = showProgress({ min: 0, max: 9, current: 0});	27	var progress = showProgress({ min: 0, max: 9, current: 0});
28		28
29	// подписываемся на события	29	// подписываемся на события
30	events.on(	30	events.on(
31	// обработчик очередного события	31	// обработчик очередного события
32	msg => {	32	msg => {
33	progress.setValue(msg);	33	progress.setValue(msg);
34	},	34	}.
35	// обработчик ошибки	35	// обработчик ошибки
36	e => {	36	e => {
37	progress.showError(e);	37	progress.showError(e);
38	},	38	},
39	// обработчик конца потока	39	// обработчик конца потока
40	() => {	40	() => {
41	progress.close();	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	```ts	51	```ts
		52	// клсс
		53	class Canvas {
		54	readonly mouseMove: IObservable<[number,number]>
		55
49	postCreate() {	56	postCreate() {
50	// превращаем события виджета в Observable	57	// превращаем события виджета в Observable
51	this.mouseMove = new Observable((notify) => {	58	this.mouseMove = new Observable<[number,number]>((notify) => {
52	this.mo~~veArea~~.on('mousemove',(x) => notify(x.) );	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	```ts	69	```ts
61		70	// класс, который будет генерировать события местоположения
62	class PositionWidget extends Widget {	71	class PositionTracker implements IDestroyable {
		72	// _nextPosition и _complete будут связаны с position при создании
		73	// экземпляра PositionTracker.
63	_nextPosition: (pos: Position) => void	74	_nextPosition: (pos: Position) => void
64
65	_complete: () => void	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	super(args);	81	super(args);
71		82
		83	// создаем Observable
72	this.position = new Observable<Position>((notify, error, complete) => {	84	this.position = new Observable<Position>((notify, error, complete) => {
		85	// сохраняем методы для оповещения о новом местоположении
73	this._nextPosition = notify;	86	this._nextPosition = notify;
		87	// метод об оповещении конца потока событий
74	this._complete = complete	88	this._complete = complete
75	});	89	});
76	}	90	}
77		91
		92	// метод для очистки ресурсов
78	destroy() {	93	destroy() {
79	this._complete();	94	this._complete();
80		95
81	super();	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	``` No newline at end of file	186	```

src/ts/Observable.ts +2 0

             import { IObservable, IDestroyable, ICancellation } from './interfaces';
             import { Cancellation } from './Cancellation'
             import { argumentNotNull } from './safe';
             interface Handler<T> {
                 (x: T): void
             }
             interface Initializer<T> {
                 (notify: Handler<T>, error?: (e: any) => void, complete?: () => void): void;
             }
             // TODO: think about to move this interfaces.ts and make it public
             interface IObserver<T> {
                 next(event: T): void
                 error(e: any): void
                 complete(): void
             }
             const noop = () => {};
             export class Observable<T> implements IObservable<T> {
                 private _once = new Array<IObserver<T>>();
                 private _observers = new Array<IObserver<T>>();
                 private _complete: boolean
                 private _error: any
                 constructor(func?: Initializer<T>) {
                     if (func)
                         func(
                             this._notifyNext.bind(this),
                             this._notifyError.bind(this),
                             this._notifyCompleted.bind(this)
                         );
                 }
                 /**
                  * Registers handlers for the current observable object.
                  *
                  * @param next the handler for events
                  * @param error the handler for a error
                  * @param complete the handler for a completion
                  * @returns {IDestroyable} the handler for the current subscription, this
                  *  handler can be used to unsubscribe from events.
                  *
                  */
                 on(next: Handler<T>, error?: Handler<any>, complete?: () => void): IDestroyable {
                     argumentNotNull(next, "next");
                     let me = this;
                     let observer: IObserver<T> & IDestroyable = {
                         next: next,
                         error: error ? error.bind(null) : noop,
                         complete: complete ? complete.bind(null) : noop,
                         destroy() {
                             me._removeObserver(this);
                         }
                     };
                     this._addObserver(observer);
                     return observer;
                 }
                 private _addObserver(observer: IObserver<T>) {
                     if (this._complete) {
                         try {
                             if (this._error)
                                 observer.error(this._error);
                             else
                                 observer.complete();
                         } catch (e) {
                             this.onObserverException(e);
                         }
                     } else {
                         this._observers.push(observer);
                     }
                 }
                 /**
                  * Waits for the next event. This method can't be used to read messages
                  * as a sequence since it can skip some messages between calls.
                  *
                  * @param ct a cancellation token
                  */
                 next(ct: ICancellation = Cancellation.none): Promise<T> {
                     return new Promise<T>((resolve, reject) => {
                         let observer: IObserver<T> = {
                             next: resolve,
                             error: reject,
                             complete: () => reject("No more events are available")
                         };
                         if (this._addOnce(observer) && ct.isSupported()) {
                             ct.register((e) => {
                                 this._removeOnce(observer);
                                 reject(e);
                             });
                         }
                     });
                 }
                 private _addOnce(observer: IObserver<T>) {
                     if (this._complete) {
                         try {
                             if (this._error)
                                 observer.error(this._error);
                             else
                                 observer.complete();
                         } catch (e) {
                             this.onObserverException(e);
                         }
                         return false;
                     }
                     this._once.push(observer);
                     return true;
                 }
                 protected onObserverException(e: any) {
                 }
                 private _removeOnce(d: IObserver<T>) {
                     let i = this._once.indexOf(d);
                     if (i >= 0)
                         this._once.splice(i, 1);
                 }
                 private _removeObserver(d: IObserver<T>) {
                     let i = this._observers.indexOf(d);
                     if (i >= 0)
                         this._observers.splice(i, 1);
                 }
                 private _notify(guard: (observer: IObserver<T>) => void) {
                     if (this._once.length) {
                         for (let i = 0; i < this._once.length; i++)
                             guard(this._once[i]);
                         this._once = [];
                     }
                     for (let i = 0; i < this._observers.length; i++)
                         guard(this._observers[i]);
                 }
                 protected _notifyNext(evt: T) {
                     let guard = (observer: IObserver<T>) => {
                         try {
                             observer.next(evt);
                         } catch (e) {
                             this.onObserverException(e);
                         }
                     }
                     this._notify(guard);
                 }
                 protected _notifyError(e: any) {
                     let guard = (observer: IObserver<T>) => {
                         try {
                             observer.error(e);
                         } catch (e) {
                             this.onObserverException(e);
                         }
                     }
                     this._notify(guard);
                     this._observers = [];
+                    this._complete = true;
                 }
                 protected _notifyCompleted() {
                     let guard = (observer: IObserver<T>) => {
                         try {
                             observer.complete();
                         } catch (e) {
                             this.onObserverException(e);
                         }
                     }
                     this._notify(guard);
                     this._observers = [];
+                    this._complete = true;
                 }
             }

test/ts/ObservableTests.ts +39 -1

             import { TraceSource, DebugLevel } from '@implab/core/log/TraceSource'
             import * as tape from 'tape';
             import { TapeWriter, delay } from './TestTraits';
             import { Observable } from '@implab/core/Observable';
             import { IObservable } from '@implab/core/interfaces';
             let trace = TraceSource.get("ObservableTests");
             tape('events sequence example', async t => {
                 let events: IObservable<number>
                 let done = new Promise<void>((resolve) => {
                     events = new Observable<number>(async (notify, fail, complete) => {
                         for (let i = 0; i < 10; i++) {
                             await delay(0);
                             notify(i);
                         }
+                        complete();
                         resolve();
                     });
                 });
                 let count = 0;
-                events.on(x => count = count + x);
+                let complete = false;
+                events.on(x => count = count + x, null, () => complete = true);
                 let first = await events.next();
                 t.equals(first, 0, "the first event");
+                t.false(complete, "the sequence is not complete");
                 await done;
                 t.equals(count, 45, "the summ of the evetns");
+                t.true(complete, "the sequence is complete");
+                t.end();
+            });
+            tape('event sequence termination', async t => {
+                let events: IObservable<number>
+                let done = new Promise<void>((resolve) => {
+                    events = new Observable<number>(async (notify, fail, complete) => {
+                        await delay(0);
+                        notify(1);
+                        complete();
+                        notify(2);
+                        complete();
+                        fail("Sequence terminated");
+                        resolve();
+                    });
+                });
+                let count = 0;
+                events.on(() => {}, (e) => count++, () => count++);
+                let first = await events.next();
+                t.equals(first, 1, "the first message");
+                try {
+                    await events.next();
+                    t.fail("shoud throw an exception");
+                } catch(e) {
+                    t.pass("the sequence is terminated");
+                }
+                await done;
+                t.equals(count, 1, "the sequence must be terminated once");
                 t.end();
             });

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages