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

Merge branch 'master' of https://github.com/waku-org/js-waku into chore/add-dispatch-to-ci

This commit is contained in:
fbarbu15 2023-08-29 20:18:53 +03:00
parent b7edff981f 39c323891c
commit a904609e19
11 changed files with 496 additions and 534 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

897
package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

4
packages/core/package.json
View File

@ -96,7 +96,7 @@
"@waku/build-utils": "*",
"chai": "^4.3.7",
"cspell": "^7.0.1",
"fast-check": "^3.8.1",
"fast-check": "^3.12.0",
"ignore-loader": "^0.1.2",
"isomorphic-fetch": "^3.0.0",
"karma": "^6.4.1",
@ -106,7 +106,7 @@
"mocha": "^10.2.0",
"npm-run-all": "^4.1.5",
"process": "^0.11.10",
"puppeteer": "^20.4.0",
"puppeteer": "^21.1.1",
"rollup": "^3.28.0",
"ts-loader": "^9.4.2",
"ts-node": "^10.9.1",

93
packages/core/src/lib/store/index.ts
View File

@ -86,6 +86,54 @@ class Store extends BaseProtocol implements IStore {
this.options = options ?? {};
}
/**
* Processes messages based on the provided callback and options.
* @private
*/
private async processMessages<T extends IDecodedMessage>(
messages: Promise<T | undefined>[],
callback: (message: T) => Promise<void | boolean> | boolean | void,
options?: QueryOptions
): Promise<boolean> {
let abort = false;
const messagesOrUndef: Array<T | undefined> = await Promise.all(messages);
let processedMessages: Array<T> = messagesOrUndef.filter(isDefined);
if (this.shouldReverseOrder(options)) {
processedMessages = processedMessages.reverse();
}
await Promise.all(
processedMessages.map(async (msg) => {
if (msg && !abort) {
abort = Boolean(await callback(msg));
}
})
);
return abort;
}
/**
* Determines whether to reverse the order of messages based on the provided options.
*
* Messages in pages are ordered from oldest (first) to most recent (last).
* https://github.com/vacp2p/rfc/issues/533
*
* @private
*/
private shouldReverseOrder(options?: QueryOptions): boolean {
return (
typeof options?.pageDirection === "undefined" ||
options?.pageDirection === PageDirection.BACKWARD
);
}
/**
* @deprecated Use `queryWithOrderedCallback` instead
**/
queryOrderedCallback = this.queryWithOrderedCallback;
/**
* Do a query to a Waku Store to retrieve historical/missed messages.
*
@ -103,42 +151,20 @@ class Store extends BaseProtocol implements IStore {
* or if an error is encountered when processing the reply,
* or if two decoders with the same content topic are passed.
*/
async queryOrderedCallback<T extends IDecodedMessage>(
async queryWithOrderedCallback<T extends IDecodedMessage>(
decoders: IDecoder<T>[],
callback: (message: T) => Promise<void | boolean> | boolean | void,
options?: QueryOptions
): Promise<void> {
let abort = false;
for await (const promises of this.queryGenerator(decoders, options)) {
if (abort) break;
const messagesOrUndef: Array<T | undefined> = await Promise.all(promises);
let messages: Array<T> = messagesOrUndef.filter(isDefined);
// Messages in pages are ordered from oldest (first) to most recent (last).
// https://github.com/vacp2p/rfc/issues/533
if (
typeof options?.pageDirection === "undefined" ||
options?.pageDirection === PageDirection.BACKWARD
) {
messages = messages.reverse();
}
await Promise.all(
messages.map(async (msg) => {
if (msg && !abort) {
abort = Boolean(await callback(msg));
}
})
);
if (await this.processMessages(promises, callback, options)) break;
}
}
/**
* Do a query to a Waku Store to retrieve historical/missed messages.
*
* The callback function takes a `Promise<WakuMessage>` in input,
* useful if messages needs to be decrypted and performance matters.
* useful if messages need to be decrypted and performance matters.
*
* The order of the messages passed to the callback is as follows:
* - within a page, messages are expected to be ordered from oldest to most recent
@ -152,7 +178,7 @@ class Store extends BaseProtocol implements IStore {
* or if an error is encountered when processing the reply,
* or if two decoders with the same content topic are passed.
*/
async queryCallbackOnPromise<T extends IDecodedMessage>(
async queryWithPromiseCallback<T extends IDecodedMessage>(
decoders: IDecoder<T>[],
callback: (
message: Promise<T | undefined>
@ -160,17 +186,15 @@ class Store extends BaseProtocol implements IStore {
options?: QueryOptions
): Promise<void> {
let abort = false;
let promises: Promise<void>[] = [];
for await (const page of this.queryGenerator(decoders, options)) {
const _promises = page.map(async (msg) => {
if (!abort) {
abort = Boolean(await callback(msg));
}
const _promises = page.map(async (msgPromise) => {
if (abort) return;
abort = Boolean(await callback(msgPromise));
});
promises = promises.concat(_promises);
await Promise.all(_promises);
if (abort) break;
}
await Promise.all(promises);
}
/**
@ -183,9 +207,6 @@ class Store extends BaseProtocol implements IStore {
* as follows:
* - within a page, messages SHOULD be ordered from oldest to most recent
* - pages direction depends on { @link QueryOptions.pageDirection }
*
* However, there is no way to guarantee the behavior of the remote node.
*
* @throws If not able to reach a Waku Store peer to query,
* or if an error is encountered when processing the reply,
* or if two decoders with the same content topic are passed.

2
packages/dns-discovery/package.json
View File

@ -59,7 +59,7 @@
"uint8arrays": "^4.0.4"
},
"devDependencies": {
"@libp2p/peer-id": "^2.0.4",
"@libp2p/peer-id": "^3.0.2",
"@libp2p/peer-id-factory": "^3.0.3",
"@multiformats/multiaddr": "^12.0.0",
"@rollup/plugin-commonjs": "^24.0.1",

4
packages/enr/package.json
View File

@ -53,7 +53,7 @@
"dependencies": {
"@ethersproject/rlp": "^5.7.0",
"@libp2p/crypto": "^1.0.17",
"@libp2p/peer-id": "^2.0.4",
"@libp2p/peer-id": "^3.0.2",
"@multiformats/multiaddr": "^12.0.0",
"@noble/secp256k1": "^1.7.1",
"@waku/utils": "0.0.10",
@ -78,7 +78,7 @@
"mocha": "^10.2.0",
"npm-run-all": "^4.1.5",
"process": "^0.11.10",
"puppeteer": "^20.4.0",
"puppeteer": "^21.1.1",
"rollup": "^3.28.0",
"ts-loader": "^9.4.2",
"typescript": "^5.0.4",

4
packages/interfaces/src/store.ts
View File

@ -46,12 +46,12 @@ export type StoreQueryOptions = {
} & ProtocolOptions;
export interface IStore extends IBaseProtocol {
queryOrderedCallback: <T extends IDecodedMessage>(
queryWithOrderedCallback: <T extends IDecodedMessage>(
decoders: IDecoder<T>[],
callback: (message: T) => Promise<void | boolean> | boolean | void,
options?: StoreQueryOptions
) => Promise<void>;
queryCallbackOnPromise: <T extends IDecodedMessage>(
queryWithPromiseCallback: <T extends IDecodedMessage>(
decoders: IDecoder<T>[],
callback: (
message: Promise<T | undefined>

4
packages/message-encryption/package.json
View File

@ -88,7 +88,7 @@
"@waku/build-utils": "*",
"chai": "^4.3.7",
"cspell": "^7.0.1",
"fast-check": "^3.8.1",
"fast-check": "^3.12.0",
"karma": "^6.4.1",
"karma-chrome-launcher": "^3.2.0",
"karma-mocha": "^2.0.1",
@ -96,7 +96,7 @@
"mocha": "^10.2.0",
"npm-run-all": "^4.1.5",
"process": "^0.11.10",
"puppeteer": "^20.4.0",
"puppeteer": "^21.1.1",
"rollup": "^3.28.0",
"ts-loader": "^9.4.2",
"typescript": "^5.0.4"

4
packages/message-hash/package.json
View File

@ -64,7 +64,7 @@
"@waku/interfaces": "0.0.17",
"chai": "^4.3.7",
"cspell": "^7.0.1",
"fast-check": "^3.7.0",
"fast-check": "^3.12.0",
"ignore-loader": "^0.1.2",
"isomorphic-fetch": "^3.0.0",
"karma": "^6.4.1",
@ -74,7 +74,7 @@
"mocha": "^10.2.0",
"npm-run-all": "^4.1.5",
"process": "^0.11.10",
"puppeteer": "^20.4.0",
"puppeteer": "^21.1.1",
"rollup": "^3.28.0",
"ts-loader": "^9.4.2",
"ts-node": "^10.9.1",

2
packages/relay/package.json
View File

@ -57,7 +57,7 @@
"@waku/utils": "0.0.10",
"chai": "^4.3.7",
"debug": "^4.3.4",
"fast-check": "^3.8.1"
"fast-check": "^3.12.0"
},
"devDependencies": {
"@rollup/plugin-commonjs": "^24.1.0",

2
packages/tests/package.json
View File

@ -51,7 +51,7 @@
},
"dependencies": {
"@libp2p/interface-compliance-tests": "^4.0.2",
"@libp2p/peer-id": "^2.0.4",
"@libp2p/peer-id": "^3.0.2",
"@waku/core": "*",
"@waku/enr": "*",
"@waku/interfaces": "*",

14
packages/tests/tests/store.node.spec.ts
View File

@ -204,7 +204,7 @@ describe("Waku Store", () => {
await waitForRemotePeer(waku, [Protocols.Store]);
const messages: IMessage[] = [];
await waku.store.queryCallbackOnPromise(
await waku.store.queryWithPromiseCallback(
[TestDecoder],
async (msgPromise) => {
const msg = await msgPromise;
@ -246,7 +246,7 @@ describe("Waku Store", () => {
const desiredMsgs = 14;
const messages: IMessage[] = [];
await waku.store.queryCallbackOnPromise(
await waku.store.queryWithPromiseCallback(
[TestDecoder],
async (msgPromise) => {
const msg = await msgPromise;
@ -285,7 +285,7 @@ describe("Waku Store", () => {
await waitForRemotePeer(waku, [Protocols.Store]);
const messages: IMessage[] = [];
await waku.store.queryOrderedCallback(
await waku.store.queryWithOrderedCallback(
[TestDecoder],
async (msg) => {
messages.push(msg);
@ -324,7 +324,7 @@ describe("Waku Store", () => {
await waitForRemotePeer(waku, [Protocols.Store]);
let messages: IMessage[] = [];
await waku.store.queryOrderedCallback(
await waku.store.queryWithOrderedCallback(
[TestDecoder],
async (msg) => {
messages.push(msg);
@ -491,7 +491,7 @@ describe("Waku Store", () => {
const nwakuPeerId = await nwaku.getPeerId();
const firstMessages: IMessage[] = [];
await waku.store.queryOrderedCallback(
await waku.store.queryWithOrderedCallback(
[TestDecoder],
(msg) => {
if (msg) {
@ -505,7 +505,7 @@ describe("Waku Store", () => {
);
const bothMessages: IMessage[] = [];
await waku.store.queryOrderedCallback(
await waku.store.queryWithOrderedCallback(
[TestDecoder],
async (msg) => {
bothMessages.push(msg);
@ -552,7 +552,7 @@ describe("Waku Store", () => {
const desiredMsgs = 14;
const messages: IMessage[] = [];
await waku.store.queryOrderedCallback(
await waku.store.queryWithOrderedCallback(
[TestDecoder],
async (msg) => {
messages.push(msg);