-
-
Notifications
You must be signed in to change notification settings - Fork 2.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor(doc-storage): operation pattern
- Loading branch information
Showing
58 changed files
with
2,226 additions
and
994 deletions.
There are no files selected for viewing
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
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 was deleted.
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| import EventEmitter2 from 'eventemitter2'; | ||
|
|
||
| import { | ||
| ConnectOp, | ||
| DestroyOp, | ||
| DisconnectOp, | ||
| OpConsumer, | ||
| type OpSubscribableHandler, | ||
| SubscribeConnectionStatusOp, | ||
| } from '../op'; | ||
| import type { Storage, StorageType } from '../storage'; | ||
|
|
||
| export class PeerStorageBackend extends OpConsumer { | ||
| private readonly storages: Map<StorageType, Storage> = new Map(); | ||
| private readonly replacedStorages: Set<Storage> = new Set(); | ||
| private readonly event = new EventEmitter2(); | ||
| private readonly storageListeners: WeakMap<Storage, () => void> = new Map(); | ||
|
|
||
| constructor(port: MessagePort) { | ||
| super(port); | ||
| this.register(ConnectOp, this.connect); | ||
| this.register(DisconnectOp, this.disconnect); | ||
| this.register(DestroyOp, this.destroy); | ||
| this.registerSubscribable( | ||
| SubscribeConnectionStatusOp, | ||
| this.onStatusChanged | ||
| ); | ||
| this.listen(); | ||
| } | ||
|
|
||
| addBackendStorage<T extends new (...args: any) => Storage>( | ||
| Impl: T, | ||
| ...opts: ConstructorParameters<T> | ||
| ) { | ||
| const storage = new Impl(...opts); | ||
| const registered = this.storages.get(storage.storageType); | ||
| if (registered) { | ||
| this.replacedStorages.add(registered); | ||
| this.storageListeners.get(registered)?.(); | ||
| } | ||
| this.storageListeners.set( | ||
| storage, | ||
| storage.connection.onStatusChanged((status, error) => { | ||
| this.event.emit('statusChanged', { | ||
| storageType: storage.storageType, | ||
| status, | ||
| error, | ||
| }); | ||
| }) | ||
| ); | ||
| this.storages.set(storage.storageType, storage); | ||
| } | ||
|
|
||
| connect = async () => { | ||
| await Promise.allSettled( | ||
| Array.from(this.storages.values()).map(async storage => { | ||
| await storage.connect(); | ||
| }) | ||
| ); | ||
|
|
||
| await Promise.allSettled( | ||
| Array.from(this.replacedStorages).map(async storage => { | ||
| await storage.disconnect(); | ||
| this.replacedStorages.delete(storage); | ||
| }) | ||
| ); | ||
| }; | ||
|
|
||
| disconnect = async () => { | ||
| await Promise.allSettled( | ||
| Object.values(this.storages).map(async storage => { | ||
| await storage.disconnect(); | ||
| }) | ||
| ); | ||
| await Promise.allSettled( | ||
| Array.from(this.replacedStorages).map(async storage => { | ||
| await storage.disconnect(); | ||
| this.replacedStorages.delete(storage); | ||
| }) | ||
| ); | ||
| }; | ||
|
|
||
| override destroy = async () => { | ||
| this.close(); | ||
| super.destroy(); | ||
| await this.disconnect(); | ||
| this.storages.clear(); | ||
| this.replacedStorages.clear(); | ||
| this.event.removeAllListeners(); | ||
| }; | ||
|
|
||
| onStatusChanged: OpSubscribableHandler<SubscribeConnectionStatusOp> = ( | ||
| _, | ||
| callback | ||
| ) => { | ||
| this.event.on('statusChanged', callback); | ||
|
|
||
| return () => { | ||
| this.event.off('statusChanged', callback); | ||
| }; | ||
| }; | ||
| } |
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,49 @@ | ||
| import { | ||
| ConnectOp, | ||
| DisconnectOp, | ||
| OpProducer, | ||
| SubscribeConnectionStatusOp, | ||
| } from '../op'; | ||
| import { PeerStorageBackend } from './backend'; | ||
|
|
||
| class PeerStorageClient extends OpProducer { | ||
| constructor( | ||
| port: MessagePort, | ||
| protected readonly backend: PeerStorageBackend | ||
| ) { | ||
| super(port); | ||
| } | ||
|
|
||
| addStorage = this.backend.addBackendStorage.bind(this.backend); | ||
|
|
||
| async connect() { | ||
| this.listen(); | ||
| await this.send(new ConnectOp()); | ||
| } | ||
|
|
||
| async disconnect() { | ||
| this.close(); | ||
| await this.send(new DisconnectOp()); | ||
| } | ||
|
|
||
| async destroy() { | ||
| await this.backend.destroy(); | ||
| this.close; | ||
| } | ||
|
|
||
| onConnectionStatusChanged() { | ||
| this.subscribe(new SubscribeConnectionStatusOp(), (/* storage */) => {}); | ||
| } | ||
| } | ||
|
|
||
| export function createPeerStorageClient() { | ||
| const channel = new MessageChannel(); | ||
| const producerPort = channel.port1; | ||
| const consumerPort = channel.port2; | ||
|
|
||
| const backend = new PeerStorageBackend(consumerPort); | ||
|
|
||
| const client = new PeerStorageClient(producerPort, backend); | ||
|
|
||
| return client; | ||
| } |
Oops, something went wrong.