-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
6 changed files
with
465 additions
and
129 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| import algosdk from 'algosdk' | ||
| import { AlgorandSubscriber } from './subscriber' | ||
| import { | ||
| getAlgodSubscribedTransactions, | ||
| getArc28EventsToProcess, | ||
| getIndexerCatchupTransactions, | ||
| prepareSubscriptionPoll, | ||
| processExtraSubscriptionTransactionFields, | ||
| } from './subscriptions' | ||
| import type { | ||
| DynamicAlgorandSubscriberConfig, | ||
| NamedTransactionFilter, | ||
| SubscribedTransaction, | ||
| TransactionSubscriptionResult, | ||
| } from './types/subscription' | ||
| import Algodv2 = algosdk.Algodv2 | ||
| import Indexer = algosdk.Indexer | ||
|
|
||
| export class DynamicAlgorandSubscriber<T> extends AlgorandSubscriber { | ||
| private pendingStateChanges: { action: 'append' | 'delete' | 'set'; stateChange: Partial<T> }[] = [] | ||
| private dynamicConfig: DynamicAlgorandSubscriberConfig<T> | ||
|
|
||
| constructor(config: DynamicAlgorandSubscriberConfig<T>, algod: Algodv2, indexer?: Indexer) { | ||
| super( | ||
| { | ||
| filters: [], | ||
| ...config, | ||
| }, | ||
| algod, | ||
| indexer, | ||
| ) | ||
| this.dynamicConfig = config | ||
| } | ||
|
|
||
| protected override async _pollOnce(watermark: number): Promise<TransactionSubscriptionResult> { | ||
| let subscribedTransactions: SubscribedTransaction[] = [] | ||
| let filterState: T = await this.dynamicConfig.filterStatePersistence.get() | ||
|
|
||
| const subscribe = async (filters: NamedTransactionFilter[]) => { | ||
| const catchupTransactions = await getIndexerCatchupTransactions(filters, pollMetadata, arc28EventsToProcess, this.indexer) | ||
| const algodTransactions = await getAlgodSubscribedTransactions(filters, pollMetadata, arc28EventsToProcess) | ||
| const subscribedTransactions = catchupTransactions | ||
| .concat(algodTransactions) | ||
| .map((t) => processExtraSubscriptionTransactionFields(t, arc28EventsToProcess, this.config.arc28Events ?? [])) | ||
| this._processFilters({ subscribedTransactions, ...pollMetadata }) | ||
| return subscribedTransactions | ||
| } | ||
|
|
||
| const filters = await this.dynamicConfig.dynamicFilters(filterState, 0) | ||
| this.filterNames = filters | ||
| .map((f) => f.name) | ||
| .filter((value, index, self) => { | ||
| // Remove duplicates | ||
| return self.findIndex((x) => x === value) === index | ||
| }) | ||
| const pollMetadata = await prepareSubscriptionPoll({ ...this.config, watermark, filters }, this.algod) | ||
| const arc28EventsToProcess = getArc28EventsToProcess(this.config.arc28Events ?? []) | ||
|
|
||
| subscribedTransactions = await subscribe(filters) | ||
|
|
||
| let pollLevel = 0 | ||
| while (this.pendingStateChanges.length > 0) { | ||
| let filterStateToProcess = { ...filterState } | ||
| for (const change of this.pendingStateChanges) { | ||
| switch (change.action) { | ||
| case 'append': | ||
| for (const key of Object.keys(change.stateChange)) { | ||
| const k = key as keyof T | ||
| if (!filterState[k] || !Array.isArray(filterState[k])) { | ||
| filterState[k] = change.stateChange[k]! | ||
| } else { | ||
| filterState[k] = (filterState[k] as unknown[]).concat(change.stateChange[k]) as T[keyof T] | ||
| } | ||
| } | ||
| filterStateToProcess = { ...filterStateToProcess, ...change.stateChange } | ||
| break | ||
| case 'delete': | ||
| for (const key of Object.keys(change.stateChange)) { | ||
| const k = key as keyof T | ||
| delete filterState[k] | ||
| delete filterStateToProcess[k] | ||
| } | ||
| break | ||
| case 'set': | ||
| filterState = { ...filterState, ...change.stateChange } | ||
| filterStateToProcess = { ...filterState, ...change.stateChange } | ||
| break | ||
| } | ||
| } | ||
| this.pendingStateChanges = [] | ||
| const newFilters = await this.dynamicConfig.dynamicFilters(filterStateToProcess, ++pollLevel) | ||
| this.filterNames = newFilters | ||
| .map((f) => f.name) | ||
| .filter((value, index, self) => { | ||
| // Remove duplicates | ||
| return self.findIndex((x) => x === value) === index | ||
| }) | ||
| subscribedTransactions = subscribedTransactions.concat(await subscribe(newFilters)) | ||
| } | ||
|
|
||
| await this.dynamicConfig.filterStatePersistence.set(filterState) | ||
|
|
||
| return { | ||
| syncedRoundRange: pollMetadata.syncedRoundRange, | ||
| newWatermark: pollMetadata.newWatermark, | ||
| currentRound: pollMetadata.currentRound, | ||
| subscribedTransactions: subscribedTransactions.sort( | ||
| (a, b) => a['confirmed-round']! - b['confirmed-round']! || a['intra-round-offset']! - b['intra-round-offset']!, | ||
| ), | ||
| } | ||
| } | ||
|
|
||
| appendFilterState(stateChange: Partial<T>) { | ||
| this.pendingStateChanges.push({ action: 'append', stateChange }) | ||
| } | ||
|
|
||
| deleteFilterState(stateChange: (keyof T)[]) { | ||
| this.pendingStateChanges.push({ | ||
| action: 'delete', | ||
| stateChange: stateChange.reduce((acc, key) => ({ ...acc, [key]: true }), {} as Partial<T>), | ||
| }) | ||
| } | ||
|
|
||
| setFilterState(stateChange: Partial<T>) { | ||
| this.pendingStateChanges.push({ action: 'set', stateChange }) | ||
| } | ||
| } |
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 |
|---|---|---|
| @@ -1,2 +1,3 @@ | ||
| export * from './subscriber' | ||
| export * from './subscriptions' | ||
| export { DynamicAlgorandSubscriber } from './dynamic-subscriber' | ||
| export { AlgorandSubscriber } from './subscriber' | ||
| export { getSubscribedTransactions } from './subscriptions' |
Oops, something went wrong.