mirror of
https://github.com/irislib/iris-messenger.git
synced 2024-10-18 06:03:22 +00:00
wip
This commit is contained in:
parent
1e3480899f
commit
811e81ea61
@ -3,7 +3,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import MemoryAdapter from '@/state/MemoryAdapter.ts';
|
||||
import { Callback, Unsubscribe } from '@/state/types.ts';
|
||||
|
||||
import Node from './Node';
|
||||
import Node, { DIR_VALUE } from './Node';
|
||||
|
||||
describe('Node', () => {
|
||||
let node;
|
||||
@ -100,10 +100,10 @@ describe('Node', () => {
|
||||
const node = new Node({ id: 'root', adapters: [new MemoryAdapter()] });
|
||||
const mockCallback: Callback = vi.fn();
|
||||
const unsubscribe = node.get('chats').map(mockCallback);
|
||||
await node.get('chats').get('someChatId').put({ id: 'someChatId' });
|
||||
await node.get('chats').get('someChatId').get('latest').put({ id: 'messageId', text: 'hi' });
|
||||
|
||||
expect(mockCallback).toHaveBeenCalledWith(
|
||||
{ id: 'someChatId' },
|
||||
DIR_VALUE,
|
||||
'root/chats/someChatId',
|
||||
expect.any(Number),
|
||||
expect.any(Function),
|
||||
|
@ -12,8 +12,13 @@ type NodeProps = {
|
||||
parent?: Node | null;
|
||||
};
|
||||
|
||||
const DIR_VALUE = '__DIR__';
|
||||
export const DIR_VALUE = '__DIR__';
|
||||
|
||||
/**
|
||||
* Nodes represent queries into the tree rather than the tree itself. The actual tree data is stored by Adapters.
|
||||
*
|
||||
* Node can be a branch node or a leaf node. Branch nodes have children, leaf nodes have a value (stored in an adapter).
|
||||
*/
|
||||
export default class Node {
|
||||
id: string;
|
||||
parent: Node | null;
|
||||
@ -21,7 +26,7 @@ export default class Node {
|
||||
on_subscriptions = new Map<number, Callback>();
|
||||
map_subscriptions = new Map<number, Callback>();
|
||||
adapters: Adapter[];
|
||||
counter = 0;
|
||||
private counter = 0;
|
||||
|
||||
constructor({ id = '', adapters, parent = null }: NodeProps = {}) {
|
||||
this.id = id;
|
||||
@ -29,6 +34,10 @@ export default class Node {
|
||||
this.adapters = adapters ?? parent?.adapters ?? [new MemoryAdapter(), new LocalForageAdapter()];
|
||||
}
|
||||
|
||||
isBranchNode() {
|
||||
return this.children.size > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @param key
|
||||
@ -45,8 +54,10 @@ export default class Node {
|
||||
return new_node;
|
||||
}
|
||||
|
||||
private async putLeaf(value: any, updatedAt: number) {
|
||||
private async putValue(value: any, updatedAt: number) {
|
||||
if (value !== DIR_VALUE) {
|
||||
this.children = new Map();
|
||||
}
|
||||
const nodeValue: NodeValue = {
|
||||
updatedAt,
|
||||
value,
|
||||
@ -58,11 +69,12 @@ export default class Node {
|
||||
await Promise.all(promises);
|
||||
}
|
||||
|
||||
private async putBranch(value: Record<string, any>, updatedAt: number) {
|
||||
private async putChildValues(value: Record<string, any>, updatedAt: number) {
|
||||
const promises = this.adapters.map((adapter) =>
|
||||
adapter.set(this.id, { value: DIR_VALUE, updatedAt }),
|
||||
);
|
||||
const children = Object.keys(value);
|
||||
// the following probably causes the same callbacks to be fired too many times
|
||||
const childPromises = children.map((key) => this.get(key).put(value[key], updatedAt));
|
||||
await Promise.all([...promises, ...childPromises]);
|
||||
}
|
||||
@ -74,36 +86,27 @@ export default class Node {
|
||||
*/
|
||||
async put(value: any, updatedAt = Date.now()) {
|
||||
if (typeof value === 'object' && value !== null) {
|
||||
await this.putBranch(value, updatedAt);
|
||||
await this.putChildValues(value, updatedAt);
|
||||
} else {
|
||||
await this.putLeaf(value, updatedAt);
|
||||
await this.putValue(value, updatedAt);
|
||||
}
|
||||
|
||||
if (this.parent) {
|
||||
this.parent.map_subscriptions.forEach((callback) => {
|
||||
callback(value, this.id, updatedAt, () => {});
|
||||
await this.parent.put(DIR_VALUE, updatedAt);
|
||||
const childName = this.id.split('/').pop()!;
|
||||
if (!this.parent.children.has(childName)) {
|
||||
this.parent.children.set(childName, this);
|
||||
}
|
||||
for (const [id, callback] of this.parent.map_subscriptions) {
|
||||
console.log('calling map callback of ', this.parent.id, ' with ', this.id, value);
|
||||
callback(value, this.id, updatedAt, () => {
|
||||
this.parent?.map_subscriptions.delete(id);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Subscribe to a value
|
||||
* @param callback
|
||||
*/
|
||||
on(callback: Callback): Unsubscribe {
|
||||
let latest: NodeValue | null = null;
|
||||
const cb = (value, path, updatedAt, unsubscribe) => {
|
||||
if (value !== DIR_VALUE && (latest === null || latest.updatedAt < value.updatedAt)) {
|
||||
latest = { value, updatedAt };
|
||||
callback(value, path, updatedAt, unsubscribe);
|
||||
// TODO send to other adapters? or PubSub which decides where to send?
|
||||
}
|
||||
};
|
||||
const subId = this.counter++;
|
||||
this.on_subscriptions.set(subId, cb);
|
||||
const adapterSubs = this.adapters.map((adapter) => adapter.get(this.id, cb));
|
||||
|
||||
if (this.children.size > 0) {
|
||||
doBranchNodeCallback(callback: Callback) {
|
||||
const aggregated: Record<string, any> = {};
|
||||
const keys = Array.from(this.children.keys());
|
||||
const total = keys.length;
|
||||
@ -121,6 +124,32 @@ export default class Node {
|
||||
});
|
||||
}
|
||||
|
||||
// note to self: may be problematic that on behaves differently for leaf and branch nodes
|
||||
/**
|
||||
* Subscribe to a value
|
||||
* @param callback
|
||||
*/
|
||||
on(callback: Callback, returnIfUndefined: boolean = false): Unsubscribe {
|
||||
let latest: NodeValue | null = null;
|
||||
const cb = (value, path, updatedAt, unsubscribe) => {
|
||||
if (value !== DIR_VALUE && (latest === null || latest.updatedAt < value.updatedAt)) {
|
||||
if (value !== undefined || returnIfUndefined) {
|
||||
latest = { value, updatedAt };
|
||||
callback(value, path, updatedAt, unsubscribe);
|
||||
}
|
||||
// TODO send to other adapters? or PubSub which decides where to send?
|
||||
}
|
||||
};
|
||||
const subId = this.counter++;
|
||||
this.on_subscriptions.set(subId, cb);
|
||||
|
||||
// if it's not a dir, adapters will call the callback directly
|
||||
const adapterSubs = this.adapters.map((adapter) => adapter.get(this.id, cb));
|
||||
|
||||
if (this.isBranchNode()) {
|
||||
this.doBranchNodeCallback(callback);
|
||||
}
|
||||
|
||||
const unsubscribe = () => {
|
||||
this.on_subscriptions.delete(subId);
|
||||
adapterSubs.forEach((unsub) => unsub());
|
||||
@ -137,7 +166,7 @@ export default class Node {
|
||||
this.map_subscriptions.set(id, callback);
|
||||
const unsubscribe = () => this.map_subscriptions.delete(id);
|
||||
for (const child of this.children.values()) {
|
||||
child.once(callback, unsubscribe);
|
||||
child.once(callback, false, unsubscribe);
|
||||
}
|
||||
return unsubscribe;
|
||||
}
|
||||
@ -147,7 +176,7 @@ export default class Node {
|
||||
* @param callback
|
||||
* @param unsubscribe
|
||||
*/
|
||||
once(callback?: Callback, unsubscribe?: Unsubscribe): Promise<any> {
|
||||
once(callback?: Callback, returnIfUndefined = false, unsubscribe?: Unsubscribe): Promise<any> {
|
||||
return new Promise((resolve) => {
|
||||
const cb = (value, updatedAt, path, unsub) => {
|
||||
if (unsubscribe) {
|
||||
@ -157,7 +186,7 @@ export default class Node {
|
||||
callback?.(value, updatedAt, path, unsub);
|
||||
unsub();
|
||||
};
|
||||
this.on(cb);
|
||||
this.on(cb, returnIfUndefined);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
@ -12,4 +12,5 @@ export type Callback = (
|
||||
export abstract class Adapter {
|
||||
abstract get(path: string, callback: Callback): Unsubscribe;
|
||||
abstract set(path: string, data: NodeValue): Promise<void>;
|
||||
// abstract list(path: string, callback: Callback): Unsubscribe; ?
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user