-
Notifications
You must be signed in to change notification settings - Fork 122
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feature(framework): new filesystem debugger view
- Loading branch information
Showing
43 changed files
with
2,440 additions
and
196 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
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,70 @@ | ||
| import { BrokerAdapter, BrokerCacheOptions, BrokerLockOptions } from '../broker.js'; | ||
| import { Type } from '@deepkit/type'; | ||
| import { ProcessLock } from '@deepkit/core'; | ||
|
|
||
| export class BrokerMemoryAdapter implements BrokerAdapter { | ||
| protected cache: { [key: string]: any } = {}; | ||
| protected channels: { [key: string]: ((m: any) => void)[] } = {}; | ||
| protected locks: { [key: string]: ProcessLock } = {}; | ||
|
|
||
| async disconnect(): Promise<void> { | ||
| } | ||
|
|
||
| async lock(id: string, options: BrokerLockOptions): Promise<void> { | ||
| const lock = new ProcessLock(id); | ||
| await lock.acquire(options.ttl, options.timeout); | ||
| this.locks[id] = lock; | ||
| } | ||
|
|
||
| async tryLock(id: string, options: BrokerLockOptions): Promise<boolean> { | ||
| const lock = new ProcessLock(id); | ||
| if (lock.tryLock(options.ttl)) { | ||
| this.locks[id] = lock; | ||
| return true; | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| async release(id: string): Promise<void> { | ||
| if (this.locks[id]) { | ||
| this.locks[id].unlock(); | ||
| delete this.locks[id]; | ||
| } | ||
| } | ||
|
|
||
| async getCache(key: string): Promise<any> { | ||
| return this.cache[key]; | ||
| } | ||
|
|
||
| async setCache(key: string, value: any, options: BrokerCacheOptions) { | ||
| this.cache[key] = value; | ||
| } | ||
|
|
||
| async increase(key: string, value: any): Promise<void> { | ||
| if (!(key in this.cache)) this.cache[key] = 0; | ||
| this.cache[key] += value; | ||
| } | ||
|
|
||
| async subscribe(key: string, callback: (message: any) => void, type: Type): Promise<{ unsubscribe: () => Promise<void> }> { | ||
| if (!(key in this.channels)) this.channels[key] = []; | ||
| const fn = (m: any) => { | ||
| callback(m); | ||
| }; | ||
| this.channels[key].push(fn); | ||
|
|
||
| return { | ||
| unsubscribe: async () => { | ||
| const index = this.channels[key].indexOf(fn); | ||
| if (index !== -1) this.channels[key].splice(index, 1); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| async publish<T>(key: string, message: T): Promise<void> { | ||
| if (!(key in this.channels)) return; | ||
| for (const callback of this.channels[key]) { | ||
| callback(message); | ||
| } | ||
| } | ||
| } | ||
|
|
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,209 @@ | ||
| import { ReceiveType, reflect, ReflectionKind, resolveReceiveType, Type } from '@deepkit/type'; | ||
| import { EventToken } from '@deepkit/event'; | ||
|
|
||
| export interface BrokerLockOptions { | ||
| /** | ||
| * Time to live in seconds. Default 2 minutes. | ||
| * | ||
| * The lock is automatically released after this time. | ||
| * This is to prevent deadlocks. | ||
| */ | ||
| ttl: number; | ||
|
|
||
| /** | ||
| * Timeout when acquiring the lock in seconds. Default 30 seconds. | ||
| * Ween a lock is not acquired after this time, an error is thrown. | ||
| */ | ||
| timeout: number; | ||
| } | ||
|
|
||
| export interface BrokerAdapter { | ||
| lock(id: string, options: BrokerLockOptions): Promise<void>; | ||
|
|
||
| tryLock(id: string, options: BrokerLockOptions): Promise<boolean>; | ||
|
|
||
| release(id: string): Promise<void>; | ||
|
|
||
| getCache(key: string, type: Type): Promise<any>; | ||
|
|
||
| setCache(key: string, value: any, options: BrokerCacheOptions, type: Type): Promise<void>; | ||
|
|
||
| increase(key: string, value: any): Promise<void>; | ||
|
|
||
| publish(key: string, message: any, type: Type): Promise<void>; | ||
|
|
||
| subscribe(key: string, callback: (message: any) => void, type: Type): Promise<{ unsubscribe: () => Promise<void> }>; | ||
|
|
||
| disconnect(): Promise<void>; | ||
| } | ||
|
|
||
| export const onBrokerLock = new EventToken('broker.lock'); | ||
|
|
||
| export interface BrokerCacheOptions { | ||
| ttl: number; | ||
| tags: string[]; | ||
| } | ||
|
|
||
| export class CacheError extends Error { | ||
| } | ||
|
|
||
| export type BrokerBusChannel<Type, Channel extends string, Parameters extends object = {}> = [Channel, Parameters, Type]; | ||
|
|
||
| export type BrokerCacheKey<Type, Key extends string, Parameters extends object = {}> = [Key, Parameters, Type]; | ||
|
|
||
| export type CacheBuilder<T extends BrokerCacheKey<any, any, any>> = (parameters: T[1], options: BrokerCacheOptions) => T[2] | Promise<T[2]>; | ||
|
|
||
| export class BrokerBus<T> { | ||
| constructor( | ||
| private channel: string, | ||
| private adapter: BrokerAdapter, | ||
| private type: Type, | ||
| ) { | ||
| } | ||
|
|
||
| async publish<T>(message: T) { | ||
| return this.adapter.publish(this.channel, message, this.type); | ||
| } | ||
|
|
||
| async subscribe(callback: (message: T) => void): Promise<{ unsubscribe: () => Promise<void> }> { | ||
| return this.adapter.subscribe(this.channel, callback, this.type); | ||
| } | ||
| } | ||
|
|
||
| export class BrokerCache<T extends BrokerCacheKey<any, any, any>> { | ||
| constructor( | ||
| private key: string, | ||
| private builder: CacheBuilder<T>, | ||
| private options: BrokerCacheOptions, | ||
| private adapter: BrokerAdapter, | ||
| private type: Type, | ||
| ) { | ||
| } | ||
|
|
||
| protected getCacheKey(parameters: T[1]): string { | ||
| //this.key contains parameters e.g. user/:id, id comes from parameters.id. let's replace all of it. | ||
| //note: we could create JIT function for this, but it's probably not worth it. | ||
| return this.key.replace(/:([a-zA-Z0-9_]+)/g, (v, name) => { | ||
| if (!(name in parameters)) throw new CacheError(`Parameter ${name} not given`); | ||
| return String(parameters[name]); | ||
| }); | ||
| } | ||
|
|
||
| async set(parameters: T[1], value: T[2], options: Partial<BrokerCacheOptions> = {}) { | ||
| const cacheKey = this.getCacheKey(parameters); | ||
| await this.adapter.setCache(cacheKey, value, { ...this.options, ...options }, this.type); | ||
| } | ||
|
|
||
| async increase(parameters: T[1], value: number) { | ||
| const cacheKey = this.getCacheKey(parameters); | ||
| await this.adapter.increase(cacheKey, value); | ||
| } | ||
|
|
||
| async get(parameters: T[1]): Promise<T[2]> { | ||
| const cacheKey = this.getCacheKey(parameters); | ||
| let entry = await this.adapter.getCache(cacheKey, this.type); | ||
| if (entry !== undefined) return entry; | ||
|
|
||
| const options: BrokerCacheOptions = { ...this.options }; | ||
| entry = await this.builder(parameters, options); | ||
| await this.adapter.setCache(cacheKey, entry, options, this.type); | ||
|
|
||
| return entry; | ||
| } | ||
| } | ||
|
|
||
| export class BrokerLock { | ||
| public acquired: boolean = false; | ||
|
|
||
| constructor( | ||
| private id: string, | ||
| private adapter: BrokerAdapter, | ||
| private options: BrokerLockOptions, | ||
| ) { | ||
| } | ||
|
|
||
| async acquire(): Promise<void> { | ||
| await this.adapter.lock(this.id, this.options); | ||
| this.acquired = true; | ||
| } | ||
|
|
||
| async try(): Promise<boolean> { | ||
| if (this.acquired) return true; | ||
|
|
||
| return this.acquired = await this.adapter.tryLock(this.id, this.options); | ||
| } | ||
|
|
||
| async release(): Promise<void> { | ||
| this.acquired = false; | ||
| await this.adapter.release(this.id); | ||
| } | ||
| } | ||
|
|
||
| export class Broker { | ||
| constructor( | ||
| private readonly adapter: BrokerAdapter | ||
| ) { | ||
| } | ||
|
|
||
| public lock(id: string, options: Partial<BrokerLockOptions> = {}): BrokerLock { | ||
| return new BrokerLock(id, this.adapter, Object.assign({ ttl: 60*2, timeout: 30 }, options)); | ||
| } | ||
|
|
||
| public disconnect(): Promise<void> { | ||
| return this.adapter.disconnect(); | ||
| } | ||
|
|
||
| protected cacheProvider: { [path: string]: (...args: any[]) => any } = {}; | ||
|
|
||
| public provideCache<T extends BrokerCacheKey<any, any, any>>(provider: (options: T[1]) => T[2] | Promise<T[2]>, type?: ReceiveType<T>) { | ||
| type = resolveReceiveType(type); | ||
| if (type.kind !== ReflectionKind.tuple) throw new CacheError(`Invalid type given`); | ||
| if (type.types[0].type.kind !== ReflectionKind.literal) throw new CacheError(`Invalid type given`); | ||
| const path = String(type.types[0].type.literal); | ||
| this.cacheProvider[path] = provider; | ||
| } | ||
|
|
||
| public cache<T extends BrokerCacheKey<any, any, any>>(type?: ReceiveType<T>): BrokerCache<T> { | ||
| type = resolveReceiveType(type); | ||
| if (type.kind !== ReflectionKind.tuple) throw new CacheError(`Invalid type given`); | ||
| if (type.types[0].type.kind !== ReflectionKind.literal) throw new CacheError(`Invalid type given`); | ||
| const path = String(type.types[0].type.literal); | ||
| const provider = this.cacheProvider[path]; | ||
| if (!provider) throw new CacheError(`No cache provider for cache ${type.typeName} (${path}) registered`); | ||
|
|
||
| return new BrokerCache<T>(path, provider, { ttl: 30, tags: [] }, this.adapter, type.types[2].type); | ||
| } | ||
|
|
||
| public async get<T>(key: string, builder: (options: BrokerCacheOptions) => Promise<T>, type?: ReceiveType<T>): Promise<T> { | ||
| if (!type) { | ||
| //type not manually provided via Broker.get<Type>, so we try to extract it from the builder. | ||
| const fn = reflect(builder); | ||
| if (fn.kind !== ReflectionKind.function) throw new CacheError(`Can not detect type of builder function`); | ||
| type = fn.return; | ||
| while (type.kind === ReflectionKind.promise) type = type.type; | ||
| } else { | ||
| type = resolveReceiveType(type); | ||
| } | ||
|
|
||
| const cache = this.adapter.getCache(key, type); | ||
| if (cache !== undefined) return cache; | ||
|
|
||
| const options: BrokerCacheOptions = { ttl: 30, tags: [] }; | ||
| const value = builder(options); | ||
| await this.adapter.setCache(key, value, options, type); | ||
| return value; | ||
| } | ||
|
|
||
| public bus<T extends BrokerBusChannel<any, any>>(type?: ReceiveType<T>): BrokerBus<T[2]> { | ||
| type = resolveReceiveType(type); | ||
| if (type.kind !== ReflectionKind.tuple) throw new CacheError(`Invalid type given`); | ||
| if (type.types[0].type.kind !== ReflectionKind.literal) throw new CacheError(`Invalid type given`); | ||
| const path = String(type.types[0].type.literal); | ||
|
|
||
| return new BrokerBus(path, this.adapter, type.types[2].type); | ||
| } | ||
|
|
||
| public queue<T>(channel: string, type?: ReceiveType<T>) { | ||
|
|
||
| } | ||
| } |
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
Oops, something went wrong.