Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

sdk: EventSubscriber: support events server #1222

Merged
merged 4 commits into from
Sep 17, 2024
Merged

sdk: EventSubscriber: support events server #1222

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bd9fe98
sdk: EventSubscriber: support events server
wphan Sep 16, 2024
322c2e1
check for eventEmitter before listening for reconnect
wphan Sep 16, 2024
5191c6a
add browser websocket support
wphan Sep 16, 2024
0aee20a
Merge remote-tracking branch 'origin/master' into wphan/events_server…
wphan Sep 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
179 changes: 125 additions & 54 deletions sdk/src/events/eventSubscriber.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,10 @@ import {
LogProvider,
EventSubscriberEvents,
WebSocketLogProviderConfig,
PollingLogProviderConfig,
EventsServerLogProviderConfig,
LogProviderType,
StreamingLogProviderConfig,
} from './types';
import { TxEventCache } from './txEventCache';
import { EventList } from './eventList';
Expand All @@ -19,6 +23,7 @@ import { EventEmitter } from 'events';
import StrictEventEmitter from 'strict-event-emitter-types';
import { getSortFn } from './sort';
import { parseLogs } from './parse';
import { EventsServerLogProvider } from './eventsServerLogProvider';

export class EventSubscriber {
private address: PublicKey;
Expand All @@ -27,6 +32,7 @@ export class EventSubscriber {
private awaitTxPromises = new Map<string, Promise<void>>();
private awaitTxResolver = new Map<string, () => void>();
private logProvider: LogProvider;
private currentProviderType: LogProviderType;
public eventEmitter: StrictEventEmitter<EventEmitter, EventSubscriberEvents>;
private lastSeenSlot: number;
private lastSeenBlockTime: number | undefined;
Expand All @@ -43,22 +49,57 @@ export class EventSubscriber {
this.eventListMap = new Map<EventType, EventList<EventType>>();
this.eventEmitter = new EventEmitter();

if (this.options.logProviderConfig.type === 'websocket') {
this.currentProviderType = this.options.logProviderConfig.type;
this.initializeLogProvider();
}

private initializeLogProvider(subscribe = false) {
if (this.currentProviderType === 'websocket') {
const logProviderConfig = this.options
.logProviderConfig as WebSocketLogProviderConfig;
this.logProvider = new WebSocketLogProvider(
// @ts-ignore
this.connection,
this.address,
this.options.commitment,
this.options.logProviderConfig.resubTimeoutMs
logProviderConfig.resubTimeoutMs
);
} else {
} else if (this.currentProviderType === 'polling') {
const logProviderConfig = this.options
.logProviderConfig as PollingLogProviderConfig;
this.logProvider = new PollingLogProvider(
// @ts-ignore
this.connection,
this.address,
options.commitment,
this.options.logProviderConfig.frequency,
this.options.logProviderConfig.batchSize
this.options.commitment,
logProviderConfig.frequency,
logProviderConfig.batchSize
);
} else if (this.currentProviderType === 'events-server') {
const logProviderConfig = this.options
.logProviderConfig as EventsServerLogProviderConfig;
this.logProvider = new EventsServerLogProvider(
logProviderConfig.url,
this.options.eventTypes,
this.options.address ? this.options.address.toString() : undefined
);
} else {
throw new Error(`Invalid log provider type: ${this.currentProviderType}`);
}

if (subscribe) {
this.logProvider.subscribe(
(txSig, slot, logs, mostRecentBlockTime, txSigIndex) => {
this.handleTxLogs(
txSig,
slot,
logs,
mostRecentBlockTime,
this.currentProviderType === 'events-server',
txSigIndex
);
},
true
);
}
}
Expand All @@ -77,6 +118,33 @@ export class EventSubscriber {
}
}

/**
* Implements fallback logic for reconnecting to LogProvider. Currently terminates at polling,
* could be improved to try the original type again after some cooldown.
*/
private updateFallbackProviderType(
reconnectAttempts: number,
maxReconnectAttempts: number
) {
if (reconnectAttempts < maxReconnectAttempts) {
return;
}

let nextProviderType = this.currentProviderType;
if (this.currentProviderType === 'events-server') {
nextProviderType = 'websocket';
} else if (this.currentProviderType === 'websocket') {
nextProviderType = 'polling';
} else if (this.currentProviderType === 'polling') {
nextProviderType = 'polling';
}

console.log(
`EventSubscriber: Failing over providerType ${this.currentProviderType} to ${nextProviderType}`
);
this.currentProviderType = nextProviderType;
}

public async subscribe(): Promise<boolean> {
try {
if (this.logProvider.isSubscribed()) {
Expand All @@ -85,52 +153,46 @@ export class EventSubscriber {

this.populateInitialEventListMap();

if (this.options.logProviderConfig.type === 'websocket') {
if (this.options.logProviderConfig.resubTimeoutMs) {
if (
this.options.logProviderConfig.maxReconnectAttempts &&
this.options.logProviderConfig.maxReconnectAttempts > 0
) {
const logProviderConfig = this.options
.logProviderConfig as WebSocketLogProviderConfig;
this.logProvider.eventEmitter.on(
'reconnect',
(reconnectAttempts) => {
if (
reconnectAttempts > logProviderConfig.maxReconnectAttempts
) {
console.log('Failing over to polling');
this.logProvider.eventEmitter.removeAllListeners('reconnect');
this.unsubscribe().then(() => {
this.logProvider = new PollingLogProvider(
// @ts-ignore
this.connection,
this.address,
this.options.commitment,
logProviderConfig.fallbackFrequency,
logProviderConfig.fallbackBatchSize
);
this.logProvider.subscribe(
(txSig, slot, logs, mostRecentBlockTime) => {
this.handleTxLogs(
txSig,
slot,
logs,
mostRecentBlockTime
);
},
true
);
});
}
if (
this.options.logProviderConfig.type === 'websocket' ||
this.options.logProviderConfig.type === 'events-server'
) {
const logProviderConfig = this.options
.logProviderConfig as StreamingLogProviderConfig;

if (this.logProvider.eventEmitter) {
this.logProvider.eventEmitter.on(
'reconnect',
async (reconnectAttempts) => {
if (reconnectAttempts > logProviderConfig.maxReconnectAttempts) {
console.log(
`EventSubscriber: Reconnect attempts ${reconnectAttempts}/${logProviderConfig.maxReconnectAttempts}, reconnecting...`
);
this.logProvider.eventEmitter.removeAllListeners('reconnect');
await this.unsubscribe();
this.updateFallbackProviderType(
reconnectAttempts,
logProviderConfig.maxReconnectAttempts
);
this.initializeLogProvider(true);
}
);
}
}
);
}
}
this.logProvider.subscribe((txSig, slot, logs, mostRecentBlockTime) => {
this.handleTxLogs(txSig, slot, logs, mostRecentBlockTime);
}, true);
this.logProvider.subscribe(
(txSig, slot, logs, mostRecentBlockTime, txSigIndex) => {
this.handleTxLogs(
txSig,
slot,
logs,
mostRecentBlockTime,
this.currentProviderType === 'events-server',
txSigIndex
);
},
true
);

return true;
} catch (e) {
Expand All @@ -144,13 +206,20 @@ export class EventSubscriber {
txSig: TransactionSignature,
slot: number,
logs: string[],
mostRecentBlockTime: number | undefined
mostRecentBlockTime: number | undefined,
fromEventsServer = false,
txSigIndex: number | undefined = undefined
): void {
if (this.txEventCache.has(txSig)) {
if (!fromEventsServer && this.txEventCache.has(txSig)) {
return;
}

const wrappedEvents = this.parseEventsFromLogs(txSig, slot, logs);
const wrappedEvents = this.parseEventsFromLogs(
txSig,
slot,
logs,
txSigIndex
);

for (const wrappedEvent of wrappedEvents) {
this.eventListMap.get(wrappedEvent.eventType).insert(wrappedEvent);
Expand Down Expand Up @@ -225,7 +294,8 @@ export class EventSubscriber {
private parseEventsFromLogs(
txSig: TransactionSignature,
slot: number,
logs: string[]
logs: string[],
txSigIndex: number | undefined
): WrappedEvents {
const records = [];
// @ts-ignore
Expand All @@ -238,7 +308,8 @@ export class EventSubscriber {
event.data.txSig = txSig;
event.data.slot = slot;
event.data.eventType = event.name;
event.data.txSigIndex = runningEventIndex;
event.data.txSigIndex =
txSigIndex !== undefined ? txSigIndex : runningEventIndex;
records.push(event.data);
}
runningEventIndex++;
Expand Down
Loading
Loading