waku-org
/
js-waku
mirror of https://github.com/waku-org/js-waku.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: filter missing messages (#2119) 

* bug: fix filter missing messages

* fix keep alive

* fix const
This commit is contained in:
Sasha 2024-09-03 00:59:21 +02:00 committed by GitHub
parent 5cfe9327b6
commit 5d3cc5fd45
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 29 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
packages/sdk/src/protocols/filter.ts
View File

@ -51,27 +51,25 @@ type ReceivedMessageHashes = {
const log = new Logger("sdk:filter"); const log = new Logger("sdk:filter");
const DEFAULT_MAX_PINGS = 3; const DEFAULT_MAX_PINGS = 2;
const DEFAULT_MAX_MISSED_MESSAGES_THRESHOLD = 3; const DEFAULT_MAX_MISSED_MESSAGES_THRESHOLD = 3;
const DEFAULT_KEEP_ALIVE = 30 * 1000; const DEFAULT_KEEP_ALIVE = 60 * 1000;
const DEFAULT_SUBSCRIBE_OPTIONS = {
keepAlive: DEFAULT_KEEP_ALIVE
};
export class SubscriptionManager implements ISubscriptionSDK { export class SubscriptionManager implements ISubscriptionSDK {
private readonly receivedMessagesHashStr: string[] = [];
private keepAliveTimer: number | null = null;
private readonly receivedMessagesHashes: ReceivedMessageHashes;
private peerFailures: Map<string, number> = new Map();
private missedMessagesByPeer: Map<string, number> = new Map();
private maxPingFailures: number = DEFAULT_MAX_PINGS;
private maxMissedMessagesThreshold = DEFAULT_MAX_MISSED_MESSAGES_THRESHOLD;
private subscribeOptions: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS;
private subscriptionCallbacks: Map< private subscriptionCallbacks: Map<
ContentTopic, ContentTopic,
SubscriptionCallback<IDecodedMessage> SubscriptionCallback<IDecodedMessage>
> = new Map(); > = new Map();
private readonly receivedMessagesHashStr: string[] = [];
private peerFailures: Map<string, number> = new Map();
private readonly receivedMessagesHashes: ReceivedMessageHashes;
private missedMessagesByPeer: Map<string, number> = new Map();
private keepAliveInterval: number = DEFAULT_KEEP_ALIVE;
private maxPingFailures: number = DEFAULT_MAX_PINGS;
private maxMissedMessagesThreshold = DEFAULT_MAX_MISSED_MESSAGES_THRESHOLD;
private keepAliveTimer: number | null = null;
public constructor( public constructor(
private readonly pubsubTopic: PubsubTopic, private readonly pubsubTopic: PubsubTopic,
@ -95,17 +93,23 @@ export class SubscriptionManager implements ISubscriptionSDK {
private addHash(hash: string, peerIdStr?: string): void { private addHash(hash: string, peerIdStr?: string): void {
this.receivedMessagesHashes.all.add(hash); this.receivedMessagesHashes.all.add(hash);
if (peerIdStr) { if (!peerIdStr) {
this.receivedMessagesHashes.nodes[peerIdStr].add(hash); return;
} }
if (!this.receivedMessagesHashes.nodes[peerIdStr]) {
this.receivedMessagesHashes.nodes[peerIdStr] = new Set();
}
this.receivedMessagesHashes.nodes[peerIdStr].add(hash);
} }
public async subscribe<T extends IDecodedMessage>( public async subscribe<T extends IDecodedMessage>(
decoders: IDecoder<T> | IDecoder<T>[], decoders: IDecoder<T> | IDecoder<T>[],
callback: Callback<T>, callback: Callback<T>,
options: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS options: SubscribeOptions = {}
): Promise<SDKProtocolResult> { ): Promise<SDKProtocolResult> {
this.keepAliveTimer = options.keepAlive || DEFAULT_KEEP_ALIVE; this.keepAliveInterval = options.keepAlive || DEFAULT_KEEP_ALIVE;
this.maxPingFailures = options.pingsBeforePeerRenewed || DEFAULT_MAX_PINGS; this.maxPingFailures = options.pingsBeforePeerRenewed || DEFAULT_MAX_PINGS;
this.maxMissedMessagesThreshold = this.maxMissedMessagesThreshold =
options.maxMissedMessagesThreshold || options.maxMissedMessagesThreshold ||
@ -154,8 +158,7 @@ export class SubscriptionManager implements ISubscriptionSDK {
this.subscriptionCallbacks.set(contentTopic, subscriptionCallback); this.subscriptionCallbacks.set(contentTopic, subscriptionCallback);
}); });
this.subscribeOptions = options; this.startSubscriptionsMaintenance(this.keepAliveInterval);
this.startSubscriptionsMaintenance(options);
return finalResult; return finalResult;
} }
@ -372,10 +375,8 @@ export class SubscriptionManager implements ISubscriptionSDK {
} }
} }
private startSubscriptionsMaintenance(options: SubscribeOptions): void { private startSubscriptionsMaintenance(interval: number): void {
if (options?.keepAlive) { this.startKeepAlivePings(interval);
this.startKeepAlivePings(options.keepAlive);
}
this.startConnectionListener(); this.startConnectionListener();
} }
@ -445,9 +446,7 @@ export class SubscriptionManager implements ISubscriptionSDK {
log.error(`networkStateListener failed to recover: ${err}`); log.error(`networkStateListener failed to recover: ${err}`);
} }
this.startKeepAlivePings( this.startKeepAlivePings(this.keepAliveInterval);
this.subscribeOptions?.keepAlive || DEFAULT_SUBSCRIBE_OPTIONS.keepAlive
);
} }
private incrementMissedMessageCount(peerIdStr: string): void { private incrementMissedMessageCount(peerIdStr: string): void {
@ -503,7 +502,7 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
* @param {IDecoder<T> | IDecoder<T>[]} decoders - A single decoder or an array of decoders to use for decoding messages. * @param {IDecoder<T> | IDecoder<T>[]} decoders - A single decoder or an array of decoders to use for decoding messages.
* @param {Callback<T>} callback - The callback function to be invoked with decoded messages. * @param {Callback<T>} callback - The callback function to be invoked with decoded messages.
* @param {ProtocolUseOptions} [protocolUseOptions] - Optional settings for using the protocol. * @param {ProtocolUseOptions} [protocolUseOptions] - Optional settings for using the protocol.
* @param {SubscribeOptions} [subscribeOptions=DEFAULT_SUBSCRIBE_OPTIONS] - Options for the subscription. * @param {SubscribeOptions} [subscribeOptions] - Options for the subscription.
* *
* @returns {Promise<SubscribeResult>} A promise that resolves to an object containing: * @returns {Promise<SubscribeResult>} A promise that resolves to an object containing:
* - subscription: The created subscription object if successful, or null if failed. * - subscription: The created subscription object if successful, or null if failed.
@ -539,7 +538,7 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
decoders: IDecoder<T> | IDecoder<T>[], decoders: IDecoder<T> | IDecoder<T>[],
callback: Callback<T>, callback: Callback<T>,
protocolUseOptions?: ProtocolUseOptions, protocolUseOptions?: ProtocolUseOptions,
subscribeOptions: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS subscribeOptions?: SubscribeOptions
): Promise<SubscribeResult> { ): Promise<SubscribeResult> {
const uniquePubsubTopics = this.getUniquePubsubTopics(decoders); const uniquePubsubTopics = this.getUniquePubsubTopics(decoders);
@ -653,7 +652,7 @@ class FilterSDK extends BaseProtocolSDK implements IFilterSDK {
public async subscribeWithUnsubscribe<T extends IDecodedMessage>( public async subscribeWithUnsubscribe<T extends IDecodedMessage>(
decoders: IDecoder<T> | IDecoder<T>[], decoders: IDecoder<T> | IDecoder<T>[],
callback: Callback<T>, callback: Callback<T>,
options: SubscribeOptions = DEFAULT_SUBSCRIBE_OPTIONS options?: SubscribeOptions
): Promise<Unsubscribe> { ): Promise<Unsubscribe> {
const uniquePubsubTopics = this.getUniquePubsubTopics<T>(decoders); const uniquePubsubTopics = this.getUniquePubsubTopics<T>(decoders);