-
Notifications
You must be signed in to change notification settings - Fork 80
/
pollingOracleAccountSubscriber.ts
125 lines (105 loc) · 3.05 KB
/
pollingOracleAccountSubscriber.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import {
DataAndSlot,
NotSubscribedError,
OracleEvents,
OracleAccountSubscriber,
} from './types';
import { Program } from '@coral-xyz/anchor';
import StrictEventEmitter from 'strict-event-emitter-types';
import { EventEmitter } from 'events';
import { PublicKey } from '@solana/web3.js';
import { BulkAccountLoader } from './bulkAccountLoader';
import { OracleClient, OraclePriceData } from '../oracles/types';
export class PollingOracleAccountSubscriber implements OracleAccountSubscriber {
isSubscribed: boolean;
program: Program;
eventEmitter: StrictEventEmitter<EventEmitter, OracleEvents>;
publicKey: PublicKey;
accountLoader: BulkAccountLoader;
oracleClient: OracleClient;
callbackId?: string;
errorCallbackId?: string;
oraclePriceData?: DataAndSlot<OraclePriceData>;
public constructor(
publicKey: PublicKey,
oracleClient: OracleClient,
accountLoader: BulkAccountLoader
) {
this.isSubscribed = false;
this.publicKey = publicKey;
this.oracleClient = oracleClient;
this.accountLoader = accountLoader;
this.eventEmitter = new EventEmitter();
}
async subscribe(): Promise<boolean> {
if (this.isSubscribed) {
return true;
}
await this.addToAccountLoader();
let subscriptionSucceeded = false;
let retries = 0;
while (!subscriptionSucceeded && retries < 5) {
await this.fetch();
subscriptionSucceeded = this.didSubscriptionSucceed();
retries++;
}
if (subscriptionSucceeded) {
this.eventEmitter.emit('update');
}
this.isSubscribed = subscriptionSucceeded;
return subscriptionSucceeded;
}
async addToAccountLoader(): Promise<void> {
if (this.callbackId) {
return;
}
this.callbackId = await this.accountLoader.addAccount(
this.publicKey,
async (buffer, slot) => {
const oraclePriceData =
await this.oracleClient.getOraclePriceDataFromBuffer(buffer);
this.oraclePriceData = { data: oraclePriceData, slot };
// @ts-ignore
this.eventEmitter.emit('oracleUpdate', oraclePriceData);
this.eventEmitter.emit('update');
}
);
this.errorCallbackId = this.accountLoader.addErrorCallbacks((error) => {
this.eventEmitter.emit('error', error);
});
}
async fetch(): Promise<void> {
await this.accountLoader.load();
const { buffer, slot } = this.accountLoader.getBufferAndSlot(
this.publicKey
);
this.oraclePriceData = {
data: await this.oracleClient.getOraclePriceDataFromBuffer(buffer),
slot,
};
}
async unsubscribe(): Promise<void> {
if (!this.isSubscribed) {
return;
}
this.accountLoader.removeAccount(this.publicKey, this.callbackId);
this.callbackId = undefined;
this.accountLoader.removeErrorCallbacks(this.errorCallbackId);
this.errorCallbackId = undefined;
this.isSubscribed = false;
}
assertIsSubscribed(): void {
if (!this.isSubscribed) {
throw new NotSubscribedError(
'You must call `subscribe` before using this function'
);
}
}
public getOraclePriceData(): DataAndSlot<OraclePriceData> {
this.assertIsSubscribed();
return this.oraclePriceData;
}
didSubscriptionSucceed(): boolean {
return !!this.oraclePriceData;
}
}