status-im/js-waku
2
0
Fork 0
mirror of https://github.com/status-im/js-waku.git synced 2025-02-20 17:08:15 +00:00
Code Issues Projects Releases Wiki Activity

Upgrade libp2p, peer-id, multiaddr

Browse Source 
Use new packages and fix most compilation errors.
Remaining error to be dealt with in waku.ts.
This commit is contained in:
Franck Royer 2022-06-20 16:48:30 +10:00
parent 6c30e8d400
commit 98c93c8283
No known key found for this signature in database
GPG Key ID: A82ED75A8DFC50A4
20 changed files with 2384 additions and 1339 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
examples/web-chat/src/command.ts
View File

@ -1,4 +1,4 @@
import { multiaddr } from "multiaddr";
import { multiaddr } from "@multiformats/multiaddr";
import PeerId from "peer-id";
import { Waku } from "js-waku";

3415
package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

10
package.json
View File

@ -68,6 +68,11 @@
"dependencies": {
"@chainsafe/libp2p-noise": "^5.0.0",
"@ethersproject/rlp": "^5.5.0",
"@libp2p/interface-peer-id": "^1.0.2",
"@libp2p/interface-peer-store": "^1.0.0",
"@libp2p/interfaces": "^3.0.2",
"@libp2p/peer-id": "^1.1.10",
"@multiformats/multiaddr": "^10.2.0",
"@noble/secp256k1": "^1.3.4",
"debug": "^4.3.1",
"dns-query": "^0.11.1",
@ -76,16 +81,13 @@
"it-length-prefixed": "^5.0.2",
"it-pipe": "^1.1.0",
"js-sha3": "^0.8.0",
"libp2p": "^0.36.2",
"libp2p": "^0.37.3",
"libp2p-bootstrap": "^0.14.0",
"libp2p-crypto": "^0.21.2",
"libp2p-gossipsub": "0.13.0",
"libp2p-interfaces": "^4.0.6",
"libp2p-mplex": "^0.10.4",
"libp2p-websockets": "^0.16.1",
"multiaddr": "^10.0.1",
"multiformats": "^9.6.5",
"peer-id": "^0.16.0",
"protons-runtime": "^1.0.4",
"uint8arrays": "^3.0.0",
"uuid": "^8.3.2",

2
src/lib/discovery/bootstrap.ts
View File

@ -1,5 +1,5 @@
import { Multiaddr } from "@multiformats/multiaddr";
import debug from "debug";
import { Multiaddr } from "multiaddr";
import { DnsNodeDiscovery, NodeCapabilityCount } from "./dns";
import { getPredefinedBootstrapNodes } from "./predefined";

2
src/lib/discovery/fetch_nodes.spec.ts
View File

@ -1,5 +1,5 @@
import { Multiaddr } from "@multiformats/multiaddr";
import { expect } from "chai";
import { Multiaddr } from "multiaddr";
import PeerId from "peer-id";
import { ENR, Waku2 } from "../enr";

2
src/lib/discovery/predefined.ts
View File

@ -1,4 +1,4 @@
import { Multiaddr } from "multiaddr";
import { Multiaddr } from "@multiformats/multiaddr";
import { getPseudoRandomSubset } from "./random_subset";

2
src/lib/enr/enr.spec.ts
View File

@ -1,5 +1,5 @@
import { Multiaddr } from "@multiformats/multiaddr";
import { assert, expect } from "chai";
import { Multiaddr } from "multiaddr";
import PeerId from "peer-id";
import { getPublicKey } from "../crypto";

38
src/lib/enr/enr.ts
View File

@ -1,9 +1,11 @@
import * as RLP from "@ethersproject/rlp";
import { Multiaddr } from "@multiformats/multiaddr";
import {
convertToBytes,
convertToString,
} from "@multiformats/multiaddr/src/convert";
import { names } from "@multiformats/multiaddr/src/protocols-table";
import debug from "debug";
import { Multiaddr, protocols } from "multiaddr";
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore: No types available
import muConvert from "multiaddr/src/convert";
import PeerId from "peer-id";
import { fromString } from "uint8arrays/from-string";
import { toString } from "uint8arrays/to-string";
@ -190,7 +192,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
get ip(): string | undefined {
const raw = this.get("ip");
if (raw) {
return muConvert.toString(protocols.names.ip4.code, raw) as string;
return convertToString(names.ip4.code, raw) as string;
} else {
return undefined;
}
@ -198,7 +200,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
set ip(ip: string | undefined) {
if (ip) {
this.set("ip", muConvert.toBytes(protocols.names.ip4.code, ip));
this.set("ip", convertToBytes(names.ip4.code, ip));
} else {
this.delete("ip");
}
@ -207,7 +209,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
get tcp(): number | undefined {
const raw = this.get("tcp");
if (raw) {
return Number(muConvert.toString(protocols.names.tcp.code, raw));
return Number(convertToString(names.tcp.code, raw));
} else {
return undefined;
}
@ -217,14 +219,14 @@ export class ENR extends Map<ENRKey, ENRValue> {
if (port === undefined) {
this.delete("tcp");
} else {
this.set("tcp", muConvert.toBytes(protocols.names.tcp.code, port));
this.set("tcp", convertToBytes(names.tcp.code, port.toString(10)));
}
}
get udp(): number | undefined {
const raw = this.get("udp");
if (raw) {
return Number(muConvert.toString(protocols.names.udp.code, raw));
return Number(convertToString(names.udp.code, raw));
} else {
return undefined;
}
@ -234,14 +236,14 @@ export class ENR extends Map<ENRKey, ENRValue> {
if (port === undefined) {
this.delete("udp");
} else {
this.set("udp", muConvert.toBytes(protocols.names.udp.code, port));
this.set("udp", convertToBytes(names.udp.code, port.toString(10)));
}
}
get ip6(): string | undefined {
const raw = this.get("ip6");
if (raw) {
return muConvert.toString(protocols.names.ip6.code, raw) as string;
return convertToString(names.ip6.code, raw) as string;
} else {
return undefined;
}
@ -249,7 +251,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
set ip6(ip: string | undefined) {
if (ip) {
this.set("ip6", muConvert.toBytes(protocols.names.ip6.code, ip));
this.set("ip6", convertToBytes(names.ip6.code, ip));
} else {
this.delete("ip6");
}
@ -258,7 +260,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
get tcp6(): number | undefined {
const raw = this.get("tcp6");
if (raw) {
return Number(muConvert.toString(protocols.names.tcp.code, raw));
return Number(convertToString(names.tcp.code, raw));
} else {
return undefined;
}
@ -268,14 +270,14 @@ export class ENR extends Map<ENRKey, ENRValue> {
if (port === undefined) {
this.delete("tcp6");
} else {
this.set("tcp6", muConvert.toBytes(protocols.names.tcp.code, port));
this.set("tcp6", convertToBytes(names.tcp.code, port.toString(10)));
}
}
get udp6(): number | undefined {
const raw = this.get("udp6");
if (raw) {
return Number(muConvert.toString(protocols.names.udp.code, raw));
return Number(convertToString(names.udp.code, raw));
} else {
return undefined;
}
@ -285,7 +287,7 @@ export class ENR extends Map<ENRKey, ENRValue> {
if (port === undefined) {
this.delete("udp6");
} else {
this.set("udp6", muConvert.toBytes(protocols.names.udp.code, port));
this.set("udp6", convertToBytes(names.udp.code, port.toString(10)));
}
}
@ -371,9 +373,9 @@ export class ENR extends Map<ENRKey, ENRValue> {
// N bytes for the ip address
// 1 or 2 bytes for the protocol as buffer (tcp or udp)
// 2 bytes for the port
const ipMa = protocols.names[isIpv6 ? "ip6" : "ip4"];
const ipMa = names[isIpv6 ? "ip6" : "ip4"];
const ipByteLen = ipMa.size / 8;
const protoMa = protocols.names[protoName];
const protoMa = names[protoName];
const protoBuf = varintEncode(protoMa.code);
const maBuf = new Uint8Array(3 + ipByteLen + protoBuf.length);
maBuf[0] = ipMa.code;

2
src/lib/enr/multiaddrs_codec.spec.ts
View File

@ -1,5 +1,5 @@
import { Multiaddr } from "@multiformats/multiaddr";
import { expect } from "chai";
import { Multiaddr } from "multiaddr";
import { decodeMultiaddrs, encodeMultiaddrs } from "./multiaddrs_codec";

2
src/lib/enr/multiaddrs_codec.ts
View File

@ -1,4 +1,4 @@
import { Multiaddr } from "multiaddr";
import { Multiaddr } from "@multiformats/multiaddr";
import { MULTIADDR_LENGTH_SIZE } from "./constants";

28
src/lib/select_peer.ts
View File

@ -1,18 +1,13 @@
import Libp2p from "libp2p";
import { Peer } from "libp2p/src/peer-store";
import { Peer } from "@libp2p/interface-peer-store";
import { Libp2p } from "libp2p";
/**
* Returns a pseudo-random peer that supports the given protocol.
* Useful for protocols such as store and light push
*/
export async function selectRandomPeer(
peersIter: AsyncIterable<Peer>
peers: Peer[]
): Promise<Peer | undefined> {
const peers = [];
for await (const peer of peersIter) {
peers.push(peer);
}
if (peers.length === 0) return;
const index = Math.round(Math.random() * (peers.length - 1));
@ -22,21 +17,18 @@ export async function selectRandomPeer(
/**
* Returns the list of peers that supports the given protocol.
*/
export async function* getPeersForProtocol(
export async function getPeersForProtocol(
libp2p: Libp2p,
protocols: string[]
): AsyncIterable<Peer> {
for await (const peer of libp2p.peerStore.getPeers()) {
let peerFound = false;
): Promise<Peer[]> {
const peers: Peer[] = [];
await libp2p.peerStore.forEach((peer) => {
for (let i = 0; i < protocols.length; i++) {
if (peer.protocols.includes(protocols[i])) {
peerFound = true;
peers.push(peer);
break;
}
}
if (!peerFound) {
continue;
}
yield peer;
}
});
return peers;
}

48
src/lib/waku.node.spec.ts
View File

@ -1,5 +1,5 @@
import type { PeerId } from "@libp2p/interface-peer-id";
import { expect } from "chai";
import PeerId from "peer-id";
import {
makeLogFileName,
@ -64,9 +64,12 @@ describe("Waku Dial [node only]", function () {
});
const connectedPeerID: PeerId = await new Promise((resolve) => {
waku.libp2p.connectionManager.on("peer:connect", (connection) => {
resolve(connection.remotePeer);
});
waku.libp2p.connectionManager.addEventListener(
"peer:connect",
(evt) => {
resolve(evt.detail.remotePeer);
}
);
});
expect(connectedPeerID.toString()).to.eq(multiAddrWithId.getPeerId());
@ -88,9 +91,12 @@ describe("Waku Dial [node only]", function () {
});
const connectedPeerID: PeerId = await new Promise((resolve) => {
waku.libp2p.connectionManager.on("peer:connect", (connection) => {
resolve(connection.remotePeer);
});
waku.libp2p.connectionManager.addEventListener(
"peer:connect",
(evt) => {
resolve(evt.detail.remotePeer);
}
);
});
const multiAddrWithId = await nwaku.getMultiaddrWithId();
@ -118,7 +124,10 @@ describe("Decryption Keys", () => {
}),
]);
waku1.addPeerToAddressBook(waku2.libp2p.peerId, waku2.libp2p.multiaddrs);
waku1.addPeerToAddressBook(
waku2.libp2p.peerId,
waku2.libp2p.getMultiaddrs()
);
await Promise.all([
waku1.waitForRemotePeer([Protocols.Relay]),
@ -247,11 +256,7 @@ describe("Wait for remote peer / get peers", function () {
await delay(1000);
await waku.waitForRemotePeer([Protocols.Store]);
const peers = [];
for await (const peer of waku.store.peers) {
peers.push(peer.id.toString());
}
const peers = (await waku.store.peers()).map((peer) => peer.id.toString());
const nimPeerId = multiAddrWithId.getPeerId();
expect(nimPeerId).to.not.be.undefined;
@ -272,10 +277,7 @@ describe("Wait for remote peer / get peers", function () {
await waku.dial(multiAddrWithId);
await waitPromise;
const peers = [];
for await (const peer of waku.store.peers) {
peers.push(peer.id.toString());
}
const peers = (await waku.store.peers()).map((peer) => peer.id.toString());
const nimPeerId = multiAddrWithId.getPeerId();
@ -295,10 +297,9 @@ describe("Wait for remote peer / get peers", function () {
await waku.dial(multiAddrWithId);
await waku.waitForRemotePeer([Protocols.LightPush]);
const peers = [];
for await (const peer of waku.lightPush.peers) {
peers.push(peer.id.toString());
}
const peers = (await waku.lightPush.peers()).map((peer) =>
peer.id.toString()
);
const nimPeerId = multiAddrWithId.getPeerId();
@ -318,10 +319,7 @@ describe("Wait for remote peer / get peers", function () {
await waku.dial(multiAddrWithId);
await waku.waitForRemotePeer([Protocols.Filter]);
const peers = [];
for await (const peer of waku.filter.peers) {
peers.push(peer.id.toString());
}
const peers = (await waku.filter.peers()).map((peer) => peer.id.toString());
const nimPeerId = multiAddrWithId.getPeerId();

11
src/lib/waku.spec.ts
View File

@ -1,5 +1,5 @@
import type { PeerId } from "@libp2p/interface-peer-id";
import { expect } from "chai";
import PeerId from "peer-id";
import { Waku } from "./waku";
@ -27,9 +27,12 @@ describe("Waku Dial", function () {
});
const connectedPeerID: PeerId = await new Promise((resolve) => {
waku.libp2p.connectionManager.on("peer:connect", (connection) => {
resolve(connection.remotePeer);
});
waku.libp2p.connectionManager.addEventListener(
"peer:connect",
(evt) => {
resolve(evt.detail.remotePeer);
}
);
});
expect(connectedPeerID).to.not.be.undefined;

47
src/lib/waku.ts
View File

@ -1,6 +1,8 @@
import { Noise } from "@chainsafe/libp2p-noise";
import type { PeerId } from "@libp2p/interface-peer-id";
import { Multiaddr, multiaddr } from "@multiformats/multiaddr";
import debug from "debug";
import Libp2p, { Connection, Libp2pModules, Libp2pOptions } from "libp2p";
import { createLibp2p, Libp2p, Libp2pOptions } from "libp2p";
import Libp2pBootstrap from "libp2p-bootstrap";
import { MuxedStream } from "libp2p-interfaces/dist/src/stream-muxer/types";
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
@ -13,8 +15,6 @@ import Websockets from "libp2p-websockets";
// @ts-ignore: No types available
import filters from "libp2p-websockets/src/filters";
import PingService from "libp2p/src/ping";
import { Multiaddr, multiaddr } from "multiaddr";
import PeerId from "peer-id";
import { Bootstrap, BootstrapOptions } from "./discovery";
import { FilterCodec, WakuFilter } from "./waku_filter";
@ -74,9 +74,7 @@ export interface CreateOptions {
* allowing its omission and letting Waku set good defaults.
* Notes that some values are overridden by {@link Waku} to ensure it implements the Waku protocol.
*/
libp2p?: Omit<Libp2pOptions & import("libp2p").CreateOptions, "modules"> & {
modules?: Partial<Libp2pModules>;
};
libp2p?: Partial<Libp2pOptions>;
/**
* Byte array used as key for the noise protocol used for connection encryption
* by [`Libp2p.create`](https://github.com/libp2p/js-libp2p/blob/master/doc/API.md#create)
@ -129,7 +127,8 @@ export class Waku {
const relayKeepAlive =
options.relayKeepAlive || DefaultRelayKeepAliveValueSecs;
libp2p.connectionManager.on("peer:connect", (connection: Connection) => {
libp2p.connectionManager.addEventListener("peer:connect", (evt) => {
const { detail: connection } = evt;
this.startKeepAlive(connection.remotePeer, pingKeepAlive, relayKeepAlive);
});
@ -144,7 +143,8 @@ export class Waku {
* >this event will **only** be triggered when the last connection is closed.
* @see https://github.com/libp2p/js-libp2p/blob/bad9e8c0ff58d60a78314077720c82ae331cc55b/doc/API.md?plain=1#L2100
*/
libp2p.connectionManager.on("peer:disconnect", (connection: Connection) => {
libp2p.connectionManager.addEventListener("peer:disconnect", (evt) => {
const { detail: connection } = evt;
this.stopKeepAlive(connection.remotePeer);
});
@ -158,11 +158,10 @@ export class Waku {
*/
static async create(options?: CreateOptions): Promise<Waku> {
// Get an object in case options or libp2p are undefined
const libp2pOpts = Object.assign({}, options?.libp2p);
// Default for Websocket filter is `all`:
// Returns all TCP and DNS based addresses, both with ws or wss.
libp2pOpts.config = Object.assign(
const libp2pOpts: Partial<Libp2pOptions> | undefined = Object.assign(
{
transport: {
[websocketsTransportKey]: {
@ -170,34 +169,20 @@ export class Waku {
},
},
},
options?.libp2p?.config
options?.libp2p
);
// TODO: Pass self-emit?
// Pass pubsub topic to relay
if (options?.pubSubTopic) {
libp2pOpts.config.pubsub = Object.assign(
{ pubSubTopic: options.pubSubTopic },
libp2pOpts.config.pubsub
);
}
libp2pOpts.modules = Object.assign({}, options?.libp2p?.modules);
libp2pOpts.pubsub = new WakuRelay({ pubSubTopic: options?.pubSubTopic });
// Default transport for libp2p is Websockets
libp2pOpts.modules = Object.assign(
{
transport: [Websockets],
},
options?.libp2p?.modules
);
libp2pOpts.transports = options?.libp2p?.transports ?? [Websockets];
// streamMuxer, connection encryption and pubsub are overridden
// as those are the only ones currently supported by Waku nodes.
libp2pOpts.modules = Object.assign(libp2pOpts.modules, {
streamMuxer: [Mplex],
connEncryption: [new Noise(options?.staticNoiseKey)],
pubsub: WakuRelay,
});
libp2pOpts.streamMuxers = [Mplex];
libp2pOpts.connectionEncryption = [new Noise(options?.staticNoiseKey)];
if (options?.bootstrap) {
const bootstrap = new Bootstrap(options?.bootstrap);
@ -225,7 +210,7 @@ export class Waku {
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore: modules property is correctly set thanks to voodoo
const libp2p = await Libp2p.create(libp2pOpts);
const libp2p = await createLibp2p(libp2pOpts);
const wakuStore = new WakuStore(libp2p, {
pubSubTopic: options?.pubSubTopic,

28
src/lib/waku_filter/index.ts
View File

@ -1,8 +1,9 @@
import type { PeerId } from "@libp2p/interface-peer-id";
import type { Peer } from "@libp2p/interface-peer-store";
import debug from "debug";
import lp from "it-length-prefixed";
import { pipe } from "it-pipe";
import Libp2p, { MuxedStream } from "libp2p";
import { Peer, PeerId } from "libp2p/src/peer-store";
import { Libp2p } from "libp2p";
import { WakuMessage as WakuMessageProto } from "../../proto/message";
import { DefaultPubSubTopic } from "../constants";
@ -104,10 +105,11 @@ export class WakuFilter {
};
}
private async onRequest({ stream }: Libp2p.HandlerProps): Promise<void> {
// `any` can be removed at the next libp2p release >0.37.3
private onRequest({ stream }: any): void {
log("Receiving message push");
try {
await pipe(
pipe(
stream.source,
lp.decode(),
async (source: AsyncIterable<Buffer>) => {
@ -184,13 +186,15 @@ export class WakuFilter {
}
}
private async newStream(peer: Peer): Promise<MuxedStream> {
const connection = this.libp2p.connectionManager.get(peer.id);
if (!connection) {
// Should be able to remove any at next libp2p release >0.37.3
private async newStream(peer: Peer): Promise<any> {
const connections = this.libp2p.connectionManager.getConnections(peer.id);
if (!connections) {
throw new Error("Failed to get a connection to the peer");
}
const { stream } = await connection.newStream(FilterCodec);
// TODO: Appropriate connection selection
const { stream } = await connections[0].newStream(FilterCodec);
return stream;
}
@ -204,7 +208,7 @@ export class WakuFilter {
);
}
} else {
peer = await this.randomPeer;
peer = await this.randomPeer();
if (!peer) {
throw new Error(
"Failed to find known peer that registers waku filter protocol"
@ -238,11 +242,11 @@ export class WakuFilter {
this.decryptionKeys.delete(hexToBytes(key));
}
get peers(): AsyncIterable<Peer> {
async peers(): Promise<Peer[]> {
return getPeersForProtocol(this.libp2p, [FilterCodec]);
}
get randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(this.peers);
async randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(await this.peers());
}
}

24
src/lib/waku_light_push/index.ts
View File

@ -1,9 +1,9 @@
import type { PeerId } from "@libp2p/interface-peer-id";
import type { Peer } from "@libp2p/interface-peer-store";
import concat from "it-concat";
import lp from "it-length-prefixed";
import { pipe } from "it-pipe";
import Libp2p from "libp2p";
import { Peer } from "libp2p/src/peer-store";
import PeerId from "peer-id";
import { Libp2p } from "libp2p";
import { PushResponse } from "../../proto/light_push";
import { DefaultPubSubTopic } from "../constants";
@ -55,16 +55,17 @@ export class WakuLightPush {
peer = await this.libp2p.peerStore.get(opts.peerId);
if (!peer) throw "Peer is unknown";
} else {
peer = await this.randomPeer;
peer = await this.randomPeer();
}
if (!peer) throw "No peer available";
if (!peer.protocols.includes(LightPushCodec))
throw "Peer does not register waku light push protocol";
const connection = this.libp2p.connectionManager.get(peer.id);
if (!connection) throw "Failed to get a connection to the peer";
const connections = this.libp2p.connectionManager.getConnections(peer.id);
if (!connections) throw "Failed to get a connection to the peer";
const { stream } = await connection.newStream(LightPushCodec);
// TODO: Appropriate connection management
const { stream } = await connections[0].newStream(LightPushCodec);
try {
const pubSubTopic = opts?.pubSubTopic
? opts.pubSubTopic
@ -97,9 +98,10 @@ export class WakuLightPush {
/**
* Returns known peers from the address book (`libp2p.peerStore`) that support
* light push protocol. Waku may or may not be currently connected to these peers.
* light push protocol. Waku may or may not be currently connected to these
* peers.
*/
get peers(): AsyncIterable<Peer> {
async peers(): Promise<Peer[]> {
return getPeersForProtocol(this.libp2p, [LightPushCodec]);
}
@ -108,7 +110,7 @@ export class WakuLightPush {
* book (`libp2p.peerStore`). Waku may or may not be currently connected to
* this peer.
*/
get randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(this.peers);
async randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(await this.peers());
}
}

20
src/lib/waku_relay/index.node.spec.ts
View File

@ -1,3 +1,4 @@
import { PeerId } from "@libp2p/interface-peer-id";
import { expect } from "chai";
import debug from "debug";
@ -45,7 +46,10 @@ describe("Waku Relay [node only]", () => {
}),
]);
log("Instances started, adding waku2 to waku1's address book");
waku1.addPeerToAddressBook(waku2.libp2p.peerId, waku2.libp2p.multiaddrs);
waku1.addPeerToAddressBook(
waku2.libp2p.peerId,
waku2.libp2p.getMultiaddrs()
);
log("Wait for mutual pubsub subscription");
await Promise.all([
@ -75,7 +79,7 @@ describe("Waku Relay [node only]", () => {
});
it("Register correct protocols", async function () {
const protocols = Array.from(waku1.libp2p.upgrader.protocols.keys());
const protocols = waku1.libp2p.registrar.getProtocols();
expect(protocols).to.contain("/vac/waku/relay/2.0.0");
expect(protocols.findIndex((value) => value.match(/sub/))).to.eq(-1);
@ -268,8 +272,14 @@ describe("Waku Relay [node only]", () => {
}),
]);
waku1.addPeerToAddressBook(waku2.libp2p.peerId, waku2.libp2p.multiaddrs);
waku3.addPeerToAddressBook(waku2.libp2p.peerId, waku2.libp2p.multiaddrs);
waku1.addPeerToAddressBook(
waku2.libp2p.peerId,
waku2.libp2p.getMultiaddrs()
);
waku3.addPeerToAddressBook(
waku2.libp2p.peerId,
waku2.libp2p.getMultiaddrs()
);
await Promise.all([
waku1.waitForRemotePeer([Protocols.Relay]),
@ -330,7 +340,7 @@ describe("Waku Relay [node only]", () => {
});
it("nwaku subscribes", async function () {
let subscribers: string[] = [];
let subscribers: PeerId[] = [];
while (subscribers.length === 0) {
await delay(200);

2
src/lib/waku_relay/index.ts
View File

@ -1,5 +1,5 @@
import debug from "debug";
import Libp2p from "libp2p";
import { Libp2p } from "libp2p";
import Gossipsub from "libp2p-gossipsub";
import { AddrInfo, MessageIdFunction } from "libp2p-gossipsub/src/interfaces";
import { MessageCache } from "libp2p-gossipsub/src/message-cache";

31
src/lib/waku_store/index.ts
View File

@ -1,13 +1,12 @@
import type { PeerId } from "@libp2p/interface-peer-id";
import { Peer } from "@libp2p/interface-peer-store";
import debug from "debug";
import concat from "it-concat";
import lp from "it-length-prefixed";
import { pipe } from "it-pipe";
import Libp2p from "libp2p";
import { Peer } from "libp2p/src/peer-store";
import PeerId from "peer-id";
import { Libp2p } from "libp2p";
import * as protoV2Beta4 from "../../proto/store_v2beta4";
import { HistoryResponse } from "../../proto/store_v2beta4";
import { DefaultPubSubTopic, StoreCodecs } from "../constants";
import { getPeersForProtocol, selectRandomPeer } from "../select_peer";
import { hexToBytes } from "../utils";
@ -15,8 +14,6 @@ import { DecryptionMethod, WakuMessage } from "../waku_message";
import { HistoryRPC, PageDirection } from "./history_rpc";
import Error = HistoryResponse.Error;
const dbg = debug("waku:store");
export const DefaultPageSize = 10;
@ -156,7 +153,7 @@ export class WakuStore {
if (!peer)
throw `Failed to retrieve connection details for provided peer in peer store: ${opts.peerId.toString()}`;
} else {
peer = await this.randomPeer;
peer = await this.randomPeer();
if (!peer)
throw "Failed to find known peer that registers waku store protocol";
}
@ -173,8 +170,9 @@ export class WakuStore {
throw `Peer does not register waku store protocol: ${peer.id.toString()}`;
Object.assign(opts, { storeCodec });
const connection = this.libp2p.connectionManager.get(peer.id);
if (!connection) throw "Failed to get a connection to the peer";
const connections = this.libp2p.connectionManager.getConnections(peer.id);
if (!connections || !connections.length)
throw "Failed to get a connection to the peer";
const decryptionKeys = Array.from(this.decryptionKeys).map(
([key, { method, contentTopics }]) => {
@ -201,10 +199,11 @@ export class WakuStore {
const messages: WakuMessage[] = [];
let cursor = undefined;
while (true) {
const { stream } = await connection.newStream(storeCodec);
// TODO: Some connection selection logic?
const { stream } = await connections[0].newStream(storeCodec);
const queryOpts = Object.assign(opts, { cursor });
const historyRpcQuery = HistoryRPC.createQuery(queryOpts);
dbg("Querying store peer", connection.remoteAddr.toString());
dbg("Querying store peer", connections[0].remoteAddr.toString());
const res = await pipe(
[historyRpcQuery.encode()],
@ -221,8 +220,8 @@ export class WakuStore {
const response = reply.response as protoV2Beta4.HistoryResponse;
if (response.error && response.error !== Error.ERROR_NONE_UNSPECIFIED) {
throw "History response contains an Error: " + response.error;
if (response.error) {
throw "History response contains an Error" + response.error;
}
if (!response.messages || !response.messages.length) {
@ -301,7 +300,7 @@ export class WakuStore {
* Returns known peers from the address book (`libp2p.peerStore`) that support
* store protocol. Waku may or may not be currently connected to these peers.
*/
get peers(): AsyncIterable<Peer> {
async peers(): Promise<Peer[]> {
const codecs = [];
for (const codec of Object.values(StoreCodecs)) {
codecs.push(codec);
@ -315,7 +314,7 @@ export class WakuStore {
* book (`libp2p.peerStore`). Waku may or may not be currently connected to
* this peer.
*/
get randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(this.peers);
async randomPeer(): Promise<Peer | undefined> {
return selectRandomPeer(await this.peers());
}
}

7
src/test_utils/nwaku.ts
View File

@ -5,10 +5,11 @@
import { ChildProcess, spawn } from "child_process";
import type { PeerId } from "@libp2p/interface-peer-id";
import { peerIdFromString } from "@libp2p/peer-id";
import { Multiaddr, multiaddr } from "@multiformats/multiaddr";
import appRoot from "app-root-path";
import debug from "debug";
import { Multiaddr, multiaddr } from "multiaddr";
import PeerId from "peer-id";
import portfinder from "portfinder";
import { DefaultPubSubTopic } from "../lib/constants";
@ -343,7 +344,7 @@ export class Nwaku {
if (!this.multiaddrWithId) throw "Nwaku did not return a ws multiaddr";
const peerIdStr = this.multiaddrWithId.getPeerId();
if (!peerIdStr) throw "Nwaku multiaddr does not contain peerId";
this.peerId = PeerId.createFromB58String(peerIdStr);
this.peerId = peerIdFromString(peerIdStr);
return { peerId: this.peerId, multiaddrWithId: this.multiaddrWithId };
}