-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #181 from m-ld/reactive-observable-query
Reactive observable query
- Loading branch information
Showing
8 changed files
with
670 additions
and
397 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| import { asapScheduler, Observable, observeOn } from 'rxjs'; | ||
| import type { GraphSubject, Iri, MeldClone, MeldReadState, MeldUpdate } from '..'; | ||
| import { updateSubject } from '..'; | ||
| import { enablePatches, produceWithPatches } from 'immer'; | ||
| import { takeUntilComplete } from '../engine/util'; | ||
|
|
||
| enablePatches(); | ||
|
|
||
| type Eventually<T> = T | undefined | Promise<T | undefined>; | ||
|
|
||
| /** | ||
| * 'Watches' the given read/follow procedures on the given m-ld clone, by | ||
| * creating an observable that emit when the initial read or update emit a | ||
| * non-null value. | ||
| * @param meld the clone to attach the observable to | ||
| * @param readValue reads an initial value; or null if not available | ||
| * @param updateValue provides an updated value, or null if not available. If | ||
| * this param is omitted, `readValue` is called for every update. | ||
| * @category Reactive | ||
| */ | ||
| export function watchQuery<T>( | ||
| meld: MeldClone, | ||
| readValue: (state: MeldReadState) => Eventually<T>, | ||
| updateValue: (update: MeldUpdate, state: MeldReadState) => Eventually<T> = | ||
| (_, state) => readValue(state) | ||
| ): Observable<T> { | ||
| return new Observable<T>(subs => { | ||
| subs.add(meld.read( | ||
| async state => { | ||
| try { | ||
| const value = await readValue(state); | ||
| !subs.closed && value != null && subs.next(value); | ||
| } catch (e) { | ||
| subs.error(e); | ||
| } | ||
| }, | ||
| async (update, state) => { | ||
| try { | ||
| const value = await updateValue(update, state); | ||
| !subs.closed && value != null && subs.next(value); | ||
| } catch (e) { | ||
| subs.error(e); | ||
| } | ||
| } | ||
| )); | ||
| }).pipe( | ||
| takeUntilComplete(meld.status), | ||
| // TODO: workaround: live lock throws due to overlapping states | ||
| observeOn(asapScheduler) | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * Shorthand for following the state of a specific subject in the m-ld clone. | ||
| * Will emit an initial state as soon as the subject exists, and every time the | ||
| * subject changes. | ||
| * @param meld the clone to attach the observable to | ||
| * @param id the subject identity IRI | ||
| * @category Reactive | ||
| */ | ||
| export function watchSubject( | ||
| meld: MeldClone, | ||
| id: Iri | ||
| ): Observable<GraphSubject> { | ||
| let subject: GraphSubject | undefined, patches = []; | ||
| return watchQuery( | ||
| meld, | ||
| async state => { | ||
| return subject = await state.get(id); | ||
| }, | ||
| async (update, state) => { | ||
| if (subject != null) { | ||
| [subject, patches] = produceWithPatches(subject, | ||
| // @ts-ignore: TS cannot cope with mutable GraphSubject | ||
| mutable => updateSubject(mutable, update)); | ||
| return patches.length ? subject : undefined; | ||
| } else { | ||
| return subject = await state.get(id); | ||
| } | ||
| } | ||
| ); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| import { clone, MeldClone } from '../src/index'; | ||
| import { MemoryLevel } from 'memory-level'; | ||
| import { MockRemotes, testConfig } from './testClones'; | ||
| import { watchQuery, watchSubject } from '../src/rx/index'; | ||
| import { take, toArray } from 'rxjs/operators'; | ||
| import { firstValueFrom } from 'rxjs'; | ||
| import { setTimeout } from 'timers/promises'; | ||
|
|
||
| describe('Rx utilities', () => { | ||
| let api: MeldClone; | ||
|
|
||
| beforeEach(async () => { | ||
| api = await clone(new MemoryLevel, MockRemotes, testConfig()); | ||
| }); | ||
|
|
||
| afterEach(() => api.close().catch(() => {/*Some tests do close*/})); | ||
|
|
||
| test('watch query emits if found sync', async () => { | ||
| const observable = watchQuery(api, () => 1, () => 2); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([1, 2]); | ||
| }); | ||
|
|
||
| test('watch query does not emit if not found on read sync', async () => { | ||
| const observable = watchQuery(api, () => undefined, () => 2); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([2]); | ||
| }); | ||
|
|
||
| test('watch query does not emit if not found on update sync', async () => { | ||
| const observable = watchQuery(api, () => 1, () => undefined); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([1]); | ||
| }); | ||
|
|
||
| test('watch query emits if found async on read', async () => { | ||
| const observable = watchQuery(api, () => setTimeout(1, 1), () => 2); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([1, 2]); | ||
| }); | ||
|
|
||
| test('watch query emits if found async on update', async () => { | ||
| const observable = watchQuery(api, | ||
| () => 1, () => setTimeout(1, 2)); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([1, 2]); | ||
| }); | ||
|
|
||
| test('watch query completes if clone closed', async () => { | ||
| const observable = watchQuery(api, () => 1, () => 2); | ||
| const found = firstValueFrom(observable.pipe(toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await api.close(); | ||
| await expect(found).resolves.toEqual([1, 2]); | ||
| }); | ||
|
|
||
| test('watch can unsubscribe between read and update', async () => { | ||
| const observable = watchQuery(api, () => 1, () => 2); | ||
| const found: number[] = []; | ||
| const subs = observable.subscribe(value => { | ||
| found.push(value); | ||
| subs.unsubscribe(); | ||
| }); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| expect(found).toEqual([1]); | ||
| }); | ||
|
|
||
| test('watch reads immediately', async () => { | ||
| const observable = watchQuery(api, state => state.get('fred')); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await expect(found).resolves.toEqual([{ '@id': 'fred', name: 'Fred' }]); | ||
| }); | ||
|
|
||
| test('watch reads on update', async () => { | ||
| const observable = watchQuery(api, state => state.get('fred')); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([{ '@id': 'fred', name: 'Fred' }]); | ||
| }); | ||
|
|
||
| test('watch subject initial state', async () => { | ||
| const observable = watchSubject(api, 'fred'); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await expect(found).resolves.toEqual([{ '@id': 'fred', name: 'Fred' }]); | ||
| }); | ||
|
|
||
| test('watch subject update', async () => { | ||
| const observable = watchSubject(api, 'fred'); | ||
| const found = firstValueFrom(observable.pipe(take(1), toArray())); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| await expect(found).resolves.toEqual([{ '@id': 'fred', name: 'Fred' }]); | ||
| }); | ||
|
|
||
| test('watch subject update', async () => { | ||
| const observable = watchSubject(api, 'fred'); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| await api.write({ '@update': { '@id': 'fred', name: 'Flintstone' } }); | ||
| await expect(found).resolves.toEqual([ | ||
| { '@id': 'fred', name: 'Fred' }, | ||
| { '@id': 'fred', name: 'Flintstone' } | ||
| ]); | ||
| }); | ||
|
|
||
| test('watched subject did not change', async () => { | ||
| const observable = watchSubject(api, 'fred'); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| // A non-overlapping update should be ignored | ||
| await api.write({ '@id': 'wilma', name: 'Wilma' } ); | ||
| await api.write({ '@update': { '@id': 'fred', name: 'Flintstone' } }); | ||
| await expect(found).resolves.toEqual([ | ||
| { '@id': 'fred', name: 'Fred' }, | ||
| { '@id': 'fred', name: 'Flintstone' } | ||
| ]); | ||
| }); | ||
|
|
||
| test('watch subject update with other data', async () => { | ||
| const observable = watchSubject(api, 'fred'); | ||
| await api.write({ '@id': 'fred', name: 'Fred' }); | ||
| const found = firstValueFrom(observable.pipe(take(2), toArray())); | ||
| // A non-overlapping update should be ignored | ||
| await api.write({ | ||
| '@id': 'wilma', | ||
| name: 'Wilma', | ||
| spouse: { '@id': 'fred', name: 'Flintstone' } | ||
| }); | ||
| await expect(found).resolves.toEqual([ | ||
| { '@id': 'fred', name: 'Fred' }, | ||
| { '@id': 'fred', name: ['Fred', 'Flintstone'] } | ||
| ]); | ||
| }); | ||
| }); |