observable.md
182 lines
| 8.0 KiB
| text/x-minidsrc
|
MarkdownLexer
andrei
|
r25 | # Observable | ||
| Универсальный способ организации потока сообщений. Данный механизм может | ||||
| использоваться для оповещения об изменениях состояний объектов или для доставки | ||||
| самостоятельных событий, например, связанных с действиями пользователя. | ||||
| Является реализацией классического шаблона наблюдателя с возможность сообщить | ||||
|
cin
|
r133 | о конце потока событий. Данная реализация не содержит никаких дополнительных | ||
|
andrei
|
r25 | функций, таких как фильтрация, канал с состоянием, преобразования сообщений и | ||
| т.п. Это сделано специально, чтобы реализация оставалась максимально простой. | ||||
| Пример того, как можно создать последовательность из 10 событий: | ||||
| ```ts | ||||
| var events = new Observable(async (notify, error, complete) => { | ||||
| // цикл в котором возникает событие | ||||
| for(let i = 0; i < 10; i++) { | ||||
| await delay(1000); | ||||
| // в качестве данных передается номер события | ||||
| notify(i); | ||||
| } | ||||
| // по окончании последовательности информируем, что событий больше не будет | ||||
|
cin
|
r133 | complete(); | ||
|
andrei
|
r25 | }); | ||
| // создаем окно с отображением хода событий | ||||
| var progress = showProgress({ min: 0, max: 9, current: 0}); | ||||
| // подписываемся на события | ||||
| events.on( | ||||
| // обработчик очередного события | ||||
| msg => { | ||||
| progress.setValue(msg); | ||||
|
cin
|
r26 | }. | ||
|
andrei
|
r25 | // обработчик ошибки | ||
| e => { | ||||
| progress.showError(e); | ||||
| }, | ||||
| // обработчик конца потока | ||||
| () => { | ||||
| progress.close(); | ||||
| } | ||||
| ); | ||||
|
cin
|
r26 | |||
| // ожидание следующего события | ||||
| let firstEvent = await events.next(); | ||||
|
andrei
|
r25 | ``` | ||
|
cin
|
r26 | `Observable` можно создавать из событий другого объекта, например, виджета: | ||
|
andrei
|
r25 | |||
| ```ts | ||||
|
cin
|
r133 | // класс | ||
|
cin
|
r26 | class Canvas { | ||
|
cin
|
r133 | mouseMove: IObservable<[number,number]>; | ||
|
cin
|
r26 | |||
| postCreate() { | ||||
| // превращаем события виджета в Observable | ||||
| this.mouseMove = new Observable<[number,number]>((notify) => { | ||||
| this.mousePad.on('mousemove',(e) => notify([e.clientX, e.clientY]) ); | ||||
| }); | ||||
| } | ||||
|
andrei
|
r25 | } | ||
| ``` | ||||
|
cin
|
r26 | Если объект инкапсулирует в себе `Observable`, он также может сохранить методы | ||
| для оповещения подписчиков для дальнейшего их использования внутри класса. | ||||
|
andrei
|
r25 | |||
| ```ts | ||||
|
cin
|
r26 | // класс, который будет генерировать события местоположения | ||
| class PositionTracker implements IDestroyable { | ||||
| // _nextPosition и _complete будут связаны с position при создании | ||||
| // экземпляра PositionTracker. | ||||
|
andrei
|
r25 | _nextPosition: (pos: Position) => void | ||
| _complete: () => void | ||||
|
cin
|
r26 | readonly position: IObservable<Position> | ||
|
andrei
|
r25 | |||
|
cin
|
r26 | // конструктор | ||
| constructor(...args: any[]) { | ||||
|
andrei
|
r25 | super(args); | ||
|
cin
|
r26 | // создаем Observable | ||
|
andrei
|
r25 | this.position = new Observable<Position>((notify, error, complete) => { | ||
|
cin
|
r26 | // сохраняем методы для оповещения о новом местоположении | ||
|
andrei
|
r25 | this._nextPosition = notify; | ||
|
cin
|
r26 | // метод об оповещении конца потока событий | ||
|
andrei
|
r25 | this._complete = complete | ||
| }); | ||||
| } | ||||
|
cin
|
r26 | // метод для очистки ресурсов | ||
|
andrei
|
r25 | destroy() { | ||
| this._complete(); | ||||
| super(); | ||||
| } | ||||
| } | ||||
|
cin
|
r26 | ``` | ||
|
andrei
|
r25 | |||
|
cin
|
r133 | Существует также несколько вариантов получения сообщений | ||
|
cin
|
r26 | |||
| ```ts | ||||
| // регистрация метода для получений событий | ||||
| let subscription = pushEvents.on((msg) => { | ||||
| displayPopup(msg); | ||||
| }); | ||||
| // подписку можно отменить, после чего обработчики больше не будут вызываться | ||||
| subcription.destroy(); | ||||
| // если требуется получить только одно сообщение можно использовать | ||||
| // асинхронный метод next(ct?: ICancellation) | ||||
| let msg = await pushEvents.next(); | ||||
| // пример метода для получения координат с карты, который использует | ||||
| // событие нажатия мышью для определения координат. | ||||
| class Map { | ||||
| /** | ||||
| Получает координаты по щелчку мыши. | ||||
| @async | ||||
| @returns [lon,lat] | ||||
| */ | ||||
| async peekCoordinates(ct: ICancellation = Cancellation.none) { | ||||
| // получаем событие клика | ||||
| let evt = this.viewport.click.next(ct); | ||||
|
cin
|
r59 | |||
|
cin
|
r26 | // преобразуем позицию на экране в координаты карты | ||
|
cin
|
r133 | return this.clientToCoordinates([evt.clientX,evt.clientY]); | ||
|
cin
|
r26 | } | ||
| } | ||||
| let map : Map; // где-то объявлено | ||||
| // пример получения координат с карты | ||||
| let coords = await map.peekCoordinates(); | ||||
| ``` | ||||
| ## Observable и последовательности | ||||
|
cin
|
r133 | Можно считать, что `Observable` это некоторая аналогия итератора только в | ||
| парадигме событийного (или реактивного) программирования. Следует также понимать, | ||||
|
cin
|
r26 | что при переходе от синхронного процедурного программирования к событийному так | ||
| же меняется и направление управления (Inverse Of Control), что означает | ||||
| следующее: | ||||
| * при работе с итераторами клиенты сами определяют момент чтения следующего | ||||
| элемента последовательности. | ||||
| * при работе с `Observable` клиенты вынуждены обрабатывать эти события по мере | ||||
| их поступления и не могут на это повлиять. | ||||
|
cin
|
r133 | Последний пункт можно изменить применив, например, буфер или канал с | ||
|
cin
|
r26 | состоянием, т.е. очередь, но данные механизмы выходят за рамки простого шаблона | ||
| наблюдателя. | ||||
| ```ts | ||||
| // обработка в цикле не гарантирует получения всех сообщений | ||||
| while(1) { | ||||
| // ожидаем следующее событие, по сути это подписка только на одно событие | ||||
| let next = await events.next(); | ||||
| // такой цикл может пропускать сообщения, поскольку асинхронная операция | ||||
| // позволит возобновить создание новых событий, на которые мы не подписаны | ||||
| await processEvent(next); | ||||
| // не только асинхронные операции могут привести к пропуску события | ||||
| // например вызов метода, который приводит к созданию события так же | ||||
| // приведет к тому, что созданное событие не будет обработано в текущем | ||||
| // цикле | ||||
| doSmthAndRiseEvent(); | ||||
| } | ||||
| // для получения всех сообщений нужно регистрировать подписчика | ||||
| events.on((data) => { | ||||
| // будет вызван для всех сообщений | ||||
| processEvent(data); | ||||
| }); | ||||
|
cin
|
r133 | ``` | ||