Add callback option to store query

CHANGELOG.md

@@ -14,6 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 - Testing: Upgrade nim-waku node to v0.3.
+- **Breaking**: Modify `WakuStore.queryHistory()` to accept one `Object` instead of multiple individual arguments.
 
 ## [0.3.0] - 2021-05-15
 

src/lib/waku_store/history_rpc.ts

@@ -3,10 +3,17 @@ import { v4 as uuid } from 'uuid';
 
 import * as proto from '../../proto/waku/v2/store';
 
+export enum Direction {
+  BACKWARD = 'backward',
+  FORWARD = 'forward',
+}
+
 export interface Options {
   contentTopics: string[];
   cursor?: proto.Index;
   pubsubTopic: string;
+  direction: Direction;
+  pageSize: number;
 }
 
 export class HistoryRPC {
@@ -16,10 +23,11 @@ export class HistoryRPC {
    * Create History Query.
    */
   static createQuery(options: Options): HistoryRPC {
+    const direction = directionToProto(options.direction);
     const pagingInfo = {
-      pageSize: 10,
+      pageSize: options.pageSize,
       cursor: options.cursor,
-      direction: proto.PagingInfo_Direction.DIRECTION_FORWARD,
+      direction,
     };
 
     const contentFilters = options.contentTopics.map((contentTopic) => {
@@ -56,3 +64,14 @@ export class HistoryRPC {
     return this.proto.response;
   }
 }
+
+function directionToProto(direction: Direction): proto.PagingInfo_Direction {
+  switch (direction) {
+    case Direction.BACKWARD:
+      return proto.PagingInfo_Direction.DIRECTION_BACKWARD_UNSPECIFIED;
+    case Direction.FORWARD:
+      return proto.PagingInfo_Direction.DIRECTION_FORWARD;
+    default:
+      return proto.PagingInfo_Direction.DIRECTION_BACKWARD_UNSPECIFIED;
+  }
+}

src/lib/waku_store/index.spec.ts

@@ -5,6 +5,8 @@ import { makeLogFileName, NimWaku, NOISE_KEY_1 } from '../../test_utils';
 import { Waku } from '../waku';
 import { DefaultContentTopic, WakuMessage } from '../waku_message';
 
+import { Direction } from './history_rpc';
+
 describe('Waku Store', () => {
   let waku: Waku;
   let nimWaku: NimWaku;
@@ -79,6 +81,7 @@ describe('Waku Store', () => {
     const messages = await waku.store.queryHistory({
       peerId: nimPeerId,
       contentTopics: [DefaultContentTopic],
+      direction: Direction.FORWARD,
     });
 
     expect(messages?.length).eq(15);

src/lib/waku_store/index.ts

@@ -7,7 +7,7 @@ import PeerId from 'peer-id';
 import { WakuMessage } from '../waku_message';
 import { DefaultPubsubTopic } from '../waku_relay';
 
-import { HistoryRPC } from './history_rpc';
+import { Direction, HistoryRPC } from './history_rpc';
 
 export const StoreCodec = '/vac/waku/store/2.0.0-beta3';
 
@@ -15,6 +15,9 @@ export interface Options {
   peerId: PeerId;
   contentTopics: string[];
   pubsubTopic?: string;
+  direction?: Direction;
+  pageSize?: number;
+  callback?: (messages: WakuMessage[]) => void;
 }
 
 /**
@@ -32,12 +35,15 @@ export class WakuStore {
    * retrieve all messages.
    * @param options.pubsubTopic The pubsub topic to retrieve. Currently, all waku nodes
    * use the same pubsub topic. This is reserved for future applications.
+   * @param options.callback Callback called on page of stored messages as they are retrieved
    * @throws If not able to reach the peer to query.
    */
   async queryHistory(options: Options): Promise<WakuMessage[] | null> {
     const opts = Object.assign(
       {
         pubsubTopic: DefaultPubsubTopic,
+        direction: Direction.BACKWARD,
+        pageSize: 10,
       },
       options
     );
@@ -55,11 +61,8 @@ export class WakuStore {
       try {
         const { stream } = await connection.newStream(StoreCodec);
         try {
-          const historyRpcQuery = HistoryRPC.createQuery({
-            contentTopics: opts.contentTopics,
-            cursor,
-            pubsubTopic: opts.pubsubTopic,
-          });
+          const queryOpts = Object.assign(opts, { cursor });
+          const historyRpcQuery = HistoryRPC.createQuery(queryOpts);
           const res = await pipe(
             [historyRpcQuery.encode()],
             lp.encode(),
@@ -82,8 +85,17 @@ export class WakuStore {
               return messages;
             }
 
-            response.messages.map((protoMsg) => {
-              messages.push(new WakuMessage(protoMsg));
+            const pageMessages = response.messages.map((protoMsg) => {
+              return new WakuMessage(protoMsg);
+            });
+
+            if (opts.callback) {
+              // TODO: Test the callback feature
+              opts.callback(pageMessages);
+            }
+
+            pageMessages.forEach((wakuMessage) => {
+              messages.push(wakuMessage);
             });
 
             const responsePageSize = response.pagingInfo?.pageSize;