Merge #118 #121 #122

118: Clean up WakuMessage API r=D4nte a=D4nte



121: Always trigger store query at connection r=D4nte a=D4nte

If no new messages are received, the rendering does not change
as dupe messages are filtered out.

122: Use provided API r=D4nte a=D4nte



Co-authored-by: Franck Royer <franck@status.im>

proto/waku/v2/message.proto

@@ -2,7 +2,7 @@ syntax = "proto3";
 
 package waku.v2;
 
-message WakuMessageProto {
+message WakuMessage {
   optional bytes payload = 1;
   optional string content_topic = 2;
   optional uint32 version = 3;

proto/waku/v2/store.proto

@@ -32,7 +32,7 @@ message HistoryQuery {
 }
 
 message HistoryResponse {
-  repeated WakuMessageProto messages = 1;
+  repeated WakuMessage messages = 1;
   PagingInfo paging_info = 2;
 }
 

src/lib/waku.ts

@@ -102,7 +102,7 @@ export default class Waku {
     await this.libp2p.dialProtocol(peer, [RelayCodec, StoreCodec]);
   }
 
-  async dialWithMultiAddr(peerId: PeerId, multiaddr: Multiaddr[]) {
+  addPeerToAddressBook(peerId: PeerId, multiaddr: Multiaddr[]) {
     this.libp2p.peerStore.addressBook.set(peerId, multiaddr);
   }
 

src/lib/waku_message.spec.ts

@@ -1,3 +1,4 @@
+import { expect } from 'chai';
 import fc from 'fast-check';
 
 import { WakuMessage } from './waku_message';
@@ -7,10 +8,10 @@ describe('Waku Message', function () {
     fc.assert(
       fc.property(fc.string(), (s) => {
         const msg = WakuMessage.fromUtf8String(s);
-        const binary = msg.toBinary();
+        const binary = msg.encode();
         const actual = WakuMessage.decode(binary);
 
-        return actual.isEqualTo(msg);
+        expect(actual).to.deep.equal(msg);
       })
     );
   });
@@ -19,7 +20,7 @@ describe('Waku Message', function () {
     fc.assert(
       fc.property(fc.string(), (s) => {
         const msg = WakuMessage.fromUtf8String(s);
-        const utf8 = msg.utf8Payload();
+        const utf8 = msg.payloadAsUtf8;
 
         return utf8 === s;
       })

src/lib/waku_message.ts

@@ -2,31 +2,30 @@
 import { Reader } from 'protobufjs/minimal';
 
 // Protecting the user from protobuf oddities
-import { WakuMessageProto } from '../proto/waku/v2/message';
+import * as proto from '../proto/waku/v2/message';
 
 export const DEFAULT_CONTENT_TOPIC = '/waku/2/default-content/proto';
 const DEFAULT_VERSION = 0;
 
 export class WakuMessage {
-  // TODO: Adopt similar design to HistoryRPC
+  public constructor(public proto: proto.WakuMessage) {}
-  private constructor(
-    public payload?: Uint8Array,
-    public contentTopic?: string,
-    public version?: number
-  ) {}
-
-  static fromProto(proto: WakuMessageProto) {
-    return new WakuMessage(proto.payload, proto.contentTopic, proto.version);
-  }
 
   /**
    * Create Message with a utf-8 string as payload
-   * @param payload
+   * @param utf8
+   * @param contentTopic
    * @returns {WakuMessage}
    */
-  static fromUtf8String(payload: string): WakuMessage {
+  static fromUtf8String(
-    const buf = Buffer.from(payload, 'utf-8');
+    utf8: string,
-    return new WakuMessage(buf, DEFAULT_CONTENT_TOPIC, DEFAULT_VERSION);
+    contentTopic: string = DEFAULT_CONTENT_TOPIC
+  ): WakuMessage {
+    const payload = Buffer.from(utf8, 'utf-8');
+    return new WakuMessage({
+      payload,
+      version: DEFAULT_VERSION,
+      contentTopic,
+    });
   }
 
   /**
@@ -39,50 +38,43 @@ export class WakuMessage {
     payload: Uint8Array,
     contentTopic: string = DEFAULT_CONTENT_TOPIC
   ): WakuMessage {
-    return new WakuMessage(payload, contentTopic, DEFAULT_VERSION);
+    return new WakuMessage({
+      payload,
+      version: DEFAULT_VERSION,
+      contentTopic,
+    });
   }
 
   static decode(bytes: Uint8Array): WakuMessage {
-    const wakuMsg = WakuMessageProto.decode(Reader.create(bytes));
+    const wakuMsg = proto.WakuMessage.decode(Reader.create(bytes));
-    return new WakuMessage(
+    return new WakuMessage(wakuMsg);
-      wakuMsg.payload,
-      wakuMsg.contentTopic,
-      wakuMsg.version
-    );
   }
 
-  toBinary(): Uint8Array {
+  encode(): Uint8Array {
-    return WakuMessageProto.encode({
+    return proto.WakuMessage.encode(this.proto).finish();
-      payload: this.payload,
-      version: this.version,
-      contentTopic: this.contentTopic,
-    }).finish();
   }
 
-  utf8Payload(): string {
+  get payloadAsUtf8(): string {
-    if (!this.payload) {
+    if (!this.proto.payload) {
       return '';
     }
 
-    return Array.from(this.payload)
+    return Array.from(this.proto.payload)
       .map((char) => {
         return String.fromCharCode(char);
       })
       .join('');
   }
 
-  // Purely for tests purposes.
+  get payload(): Uint8Array | undefined {
-  // We do consider protobuf field when checking equality
+    return this.proto.payload;
-  // As the content is held by the other fields.
+  }
-  isEqualTo(other: WakuMessage) {
+
-    const payloadsAreEqual =
+  get contentTopic(): string | undefined {
-      this.payload && other.payload
+    return this.proto.contentTopic;
-        ? Buffer.compare(this.payload, other.payload) === 0
+  }
-        : !(this.payload || other.payload);
+
-    return (
+  get version(): number | undefined {
-      payloadsAreEqual &&
+    return this.proto.version;
-      this.contentTopic === other.contentTopic &&
-      this.version === other.version
-    );
   }
 }

src/lib/waku_relay/index.spec.ts

@@ -32,7 +32,10 @@ describe('Waku Relay', () => {
       }),
     ]);
 
-    await waku1.dialWithMultiAddr(waku2.libp2p.peerId, waku2.libp2p.multiaddrs);
+    await waku1.addPeerToAddressBook(
+      waku2.libp2p.peerId,
+      waku2.libp2p.multiaddrs
+    );
 
     await Promise.all([
       new Promise((resolve) =>
@@ -319,7 +322,7 @@ describe('Waku Relay', () => {
         console.log('Waiting for message');
         const waku2ReceivedMsg = await waku2ReceivedPromise;
 
-        expect(waku2ReceivedMsg.utf8Payload()).to.eq(msgStr);
+        expect(waku2ReceivedMsg.payloadAsUtf8).to.eq(msgStr);
       });
     });
   });

src/lib/waku_relay/index.ts

@@ -90,7 +90,7 @@ export class WakuRelay extends Gossipsub {
    * @returns {Promise<void>}
    */
   async send(message: WakuMessage) {
-    const msg = message.toBinary();
+    const msg = message.encode();
     await super.publish(constants.RelayDefaultTopic, Buffer.from(msg));
   }
 

src/lib/waku_store/index.spec.ts

@@ -41,7 +41,7 @@ describe('Waku Store', () => {
 
     expect(messages?.length).eq(2);
     const result = messages?.findIndex((msg) => {
-      return msg.utf8Payload() === 'Message 0';
+      return msg.payloadAsUtf8 === 'Message 0';
     });
     expect(result).to.not.eq(-1);
   });
@@ -75,7 +75,7 @@ describe('Waku Store', () => {
     for (let index = 0; index < 2; index++) {
       expect(
         messages?.findIndex((msg) => {
-          return msg.utf8Payload() === `Message ${index}`;
+          return msg.payloadAsUtf8 === `Message ${index}`;
         })
       ).to.eq(index);
     }

src/lib/waku_store/index.ts

@@ -65,7 +65,7 @@ export class WakuStore {
             }
 
             response.messages.map((protoMsg) => {
-              messages.push(WakuMessage.fromProto(protoMsg));
+              messages.push(new WakuMessage(protoMsg));
             });
 
             const responsePageSize = response.pagingInfo?.pageSize;

src/proto/waku/v2/message.ts

@@ -4,18 +4,18 @@ import _m0 from 'protobufjs/minimal';
 
 export const protobufPackage = 'waku.v2';
 
-export interface WakuMessageProto {
+export interface WakuMessage {
   payload?: Uint8Array | undefined;
   contentTopic?: string | undefined;
   version?: number | undefined;
   timestamp?: number | undefined;
 }
 
-const baseWakuMessageProto: object = {};
+const baseWakuMessage: object = {};
 
-export const WakuMessageProto = {
+export const WakuMessage = {
   encode(
-    message: WakuMessageProto,
+    message: WakuMessage,
     writer: _m0.Writer = _m0.Writer.create()
   ): _m0.Writer {
     if (message.payload !== undefined) {
@@ -33,10 +33,10 @@ export const WakuMessageProto = {
     return writer;
   },
 
-  decode(input: _m0.Reader | Uint8Array, length?: number): WakuMessageProto {
+  decode(input: _m0.Reader | Uint8Array, length?: number): WakuMessage {
     const reader = input instanceof _m0.Reader ? input : new _m0.Reader(input);
     let end = length === undefined ? reader.len : reader.pos + length;
-    const message = { ...baseWakuMessageProto } as WakuMessageProto;
+    const message = { ...baseWakuMessage } as WakuMessage;
     while (reader.pos < end) {
       const tag = reader.uint32();
       switch (tag >>> 3) {
@@ -60,8 +60,8 @@ export const WakuMessageProto = {
     return message;
   },
 
-  fromJSON(object: any): WakuMessageProto {
+  fromJSON(object: any): WakuMessage {
-    const message = { ...baseWakuMessageProto } as WakuMessageProto;
+    const message = { ...baseWakuMessage } as WakuMessage;
     if (object.payload !== undefined && object.payload !== null) {
       message.payload = bytesFromBase64(object.payload);
     }
@@ -83,7 +83,7 @@ export const WakuMessageProto = {
     return message;
   },
 
-  toJSON(message: WakuMessageProto): unknown {
+  toJSON(message: WakuMessage): unknown {
     const obj: any = {};
     message.payload !== undefined &&
       (obj.payload =
@@ -97,8 +97,8 @@ export const WakuMessageProto = {
     return obj;
   },
 
-  fromPartial(object: DeepPartial<WakuMessageProto>): WakuMessageProto {
+  fromPartial(object: DeepPartial<WakuMessage>): WakuMessage {
-    const message = { ...baseWakuMessageProto } as WakuMessageProto;
+    const message = { ...baseWakuMessage } as WakuMessage;
     if (object.payload !== undefined && object.payload !== null) {
       message.payload = object.payload;
     } else {

src/proto/waku/v2/store.ts

@@ -1,7 +1,7 @@
 /* eslint-disable */
 import Long from 'long';
 import _m0 from 'protobufjs/minimal';
-import { WakuMessageProto } from '../../waku/v2/message';
+import { WakuMessage } from '../../waku/v2/message';
 
 export const protobufPackage = 'waku.v2';
 
@@ -65,7 +65,7 @@ export interface HistoryQuery {
 }
 
 export interface HistoryResponse {
-  messages: WakuMessageProto[];
+  messages: WakuMessage[];
   pagingInfo: PagingInfo | undefined;
 }
 
@@ -453,7 +453,7 @@ export const HistoryResponse = {
     writer: _m0.Writer = _m0.Writer.create()
   ): _m0.Writer {
     for (const v of message.messages) {
-      WakuMessageProto.encode(v!, writer.uint32(10).fork()).ldelim();
+      WakuMessage.encode(v!, writer.uint32(10).fork()).ldelim();
     }
     if (message.pagingInfo !== undefined) {
       PagingInfo.encode(message.pagingInfo, writer.uint32(18).fork()).ldelim();
@@ -470,9 +470,7 @@ export const HistoryResponse = {
       const tag = reader.uint32();
       switch (tag >>> 3) {
         case 1:
-          message.messages.push(
+          message.messages.push(WakuMessage.decode(reader, reader.uint32()));
-            WakuMessageProto.decode(reader, reader.uint32())
-          );
           break;
         case 2:
           message.pagingInfo = PagingInfo.decode(reader, reader.uint32());
@@ -490,7 +488,7 @@ export const HistoryResponse = {
     message.messages = [];
     if (object.messages !== undefined && object.messages !== null) {
       for (const e of object.messages) {
-        message.messages.push(WakuMessageProto.fromJSON(e));
+        message.messages.push(WakuMessage.fromJSON(e));
       }
     }
     if (object.pagingInfo !== undefined && object.pagingInfo !== null) {
@@ -505,7 +503,7 @@ export const HistoryResponse = {
     const obj: any = {};
     if (message.messages) {
       obj.messages = message.messages.map((e) =>
-        e ? WakuMessageProto.toJSON(e) : undefined
+        e ? WakuMessage.toJSON(e) : undefined
       );
     } else {
       obj.messages = [];
@@ -522,7 +520,7 @@ export const HistoryResponse = {
     message.messages = [];
     if (object.messages !== undefined && object.messages !== null) {
       for (const e of object.messages) {
-        message.messages.push(WakuMessageProto.fromPartial(e));
+        message.messages.push(WakuMessage.fromPartial(e));
       }
     }
     if (object.pagingInfo !== undefined && object.pagingInfo !== null) {

web-chat/src/App.tsx

@@ -86,7 +86,7 @@ export default function App() {
     } else {
       stateWaku.libp2p.pubsub.on(RelayDefaultTopic, handleNewMessages);
 
-      stateWaku.libp2p.peerStore.once(
+      stateWaku.libp2p.peerStore.on(
         'change:protocols',
         handleProtocolChange.bind({}, stateWaku)
       );

web-chat/src/command.ts

@@ -42,7 +42,7 @@ function connect(peer: string | undefined, waku: Waku | undefined): string[] {
     if (!peerId) {
       return ['Peer Id needed to dial'];
     }
-    waku.libp2p.peerStore.addressBook.add(PeerId.createFromB58String(peerId), [
+    waku.addPeerToAddressBook(PeerId.createFromB58String(peerId), [
       peerMultiaddr,
     ]);
     return [