Merge pull request #32 from EOSIO/upgrade-demux-3

Upgrade to demux v3

package.json

@@ -1,6 +1,6 @@
 {
   "name": "demux-eos",
-  "version": "2.0.0",
+  "version": "3.0.0",
   "description": "Demux-js Action Reader implementations for EOSIO blockchains",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
@@ -29,13 +29,13 @@
   },
   "dependencies": {
     "bunyan": "^1.8.12",
-    "demux": "^2.0.0",
+    "demux": "^3.1.3",
     "mongodb": "^3.1.3",
     "request": "^2.87.0",
     "request-promise-native": "^1.0.5"
   },
   "peerDependencies": {
-    "demux": "^2.0.0"
+    "demux": "^3.1.3"
   },
   "scripts": {
     "release": "release-it",

src/MongoActionReader.test.ts

@@ -1,7 +1,7 @@
 import { MongoActionReader } from "./MongoActionReader"
 import { mockConnect } from "./testHelpers/mongoMock"
+import { MongoClient } from "mongodb"
 
-const { MongoClient } = require.requireActual("mongodb")
 MongoClient.connect = jest.fn(() => mockConnect)
 
 describe("MongoActionReader", () => {
@@ -12,12 +12,17 @@ describe("MongoActionReader", () => {
     await reader.initialize()
   })
 
-  it("getHeadBlockNumber returns the correct blockNumber", async () => {
+  it("returns the head block number", async () => {
     const blockNum = await reader.getHeadBlockNumber()
     expect(blockNum).toEqual(21)
   })
 
-  it("getBlock returns expected block", async () => {
+  it("returns the last irreversible block number", async () => {
+    const blockNum = await reader.getLastIrreversibleBlockNumber()
+    expect(blockNum).toEqual(20)
+  })
+
+  it("returns block with the expected block number", async () => {
     const returnedBlock = await reader.getBlock(21)
     expect(returnedBlock.blockInfo.blockNumber).toEqual(21)
   })

src/MongoActionReader.ts

@@ -14,7 +14,7 @@ function wait(ms: number) {
  */
 export class MongoActionReader extends AbstractActionReader {
   private mongodb: Db | null
-  private log: Logger
+  protected log: Logger
   constructor(
     protected mongoEndpoint: string = "mongodb://127.0.0.1:27017",
     public startAtBlock: number = 1,
@@ -44,10 +44,6 @@ export class MongoActionReader extends AbstractActionReader {
           .sort({ $natural: -1 })
           .toArray()
 
-        if (this.onlyIrreversible) {
-          return blockInfo.block_header_state.dpos_irreversible_blocknum
-        }
-
         return blockInfo.block_header_state.block_num
       } catch (err) {
         if (numTries - 1 === numRetries) {
@@ -61,6 +57,31 @@ export class MongoActionReader extends AbstractActionReader {
     throw Error("Unknown error getting head block number.")
   }
 
+  public async getLastIrreversibleBlockNumber(numRetries: number = 120, waitTimeMs: number = 250): Promise<number> {
+    this.throwIfNotInitialized()
+
+    let numTries = 1
+    while (numTries <= numRetries + 1) {
+      try {
+        const [blockInfo] = await this.mongodb!.collection("block_states")
+          .find({})
+          .limit(1)
+          .sort({ $natural: -1 })
+          .toArray()
+
+          return blockInfo.block_header_state.dpos_irreversible_blocknum
+      } catch (err) {
+        if (numTries - 1 === numRetries) {
+          throw err
+        }
+        this.log.error("error getting last irreversible block number, retrying...")
+      }
+      numTries += 1
+      await wait(waitTimeMs)
+    }
+    throw Error("Unknown error getting last irreversible block number.")
+  }
+
   public async getBlock(blockNumber: number, numRetries: number = 120, waitTimeMs: number = 250): Promise<MongoBlock> {
     this.throwIfNotInitialized()
 

src/NodeosActionReader.test.ts

@@ -1,8 +1,8 @@
 import { NodeosActionReader } from "./NodeosActionReader"
 import { nodeosRawBlock } from "./testHelpers/nodeosRawBlock"
+import request from "request-promise-native"
 
 describe("NodeosActionReader", () => {
-  let request: any
   let reader: NodeosActionReader
 
   const blockInfo = {
@@ -11,74 +11,27 @@ describe("NodeosActionReader", () => {
   }
 
   beforeAll(() => {
-    request = {
-      get: async () => blockInfo,
-      post: async () => nodeosRawBlock,
-    }
+      request.get = jest.fn(async () => blockInfo)
+      request.post = jest.fn(async () => nodeosRawBlock)
   })
 
   beforeEach(() => {
-    reader = new NodeosActionReader("", 1, false, 600, request)
+    reader = new NodeosActionReader("", 10, false, 600)
   })
 
-  it("returns last irreversible block if configured", async () => {
-    reader = new NodeosActionReader("", 1, true, 600, request)
+  it("returns head block number", async () => {
     const blockNum = await reader.getHeadBlockNumber()
-    expect(blockNum).toBe(10)
+    expect(blockNum).toBe(20)
   })
 
-  it("returns head block if configured", async () => {
-    const blockNum = await reader.getHeadBlockNumber()
-    expect(blockNum).toBe(20)
+
+  it("returns last irreversible block number", async () => {
+    const blockNum = await reader.getLastIrreversibleBlockNumber()
+    expect(blockNum).toBe(10)
   })
 
-  it("gets a block", async () => {
+  it("gets block with correct block number", async () => {
     const block = await reader.getBlock(20)
-    expect(block).toEqual({
-      actions: [
-        {
-          payload: {
-            account: "testing",
-            actionIndex: 0,
-            authorization: [
-              {
-                actor: "testing",
-                permission: "active",
-              },
-            ],
-            data: {
-              memo: "EOS is awesome!",
-            },
-            name: "action",
-            transactionId: "b890beb84a6d1d77755f2e0cdad48e2ffcfd06ff3481917b4875cc5f3a343533",
-          },
-          type: "testing::action",
-        },
-        {
-          payload: {
-            account: "testing",
-            actionIndex: 1,
-            authorization: [
-              {
-                actor: "testing",
-                permission: "active",
-              },
-            ],
-            data: {
-              memo: "Go EOS!",
-            },
-            name: "action2",
-            transactionId: "b890beb84a6d1d77755f2e0cdad48e2ffcfd06ff3481917b4875cc5f3a343533",
-          },
-          type: "testing::action2",
-        },
-      ],
-      blockInfo: {
-        blockHash: "000f4241873a9aef0daefd47d8821495b6f61c4d1c73544419eb0ddc22a9e906",
-        blockNumber: 20,
-        previousBlockHash: "000f42401b5636c3c1d88f31fe0e503654091fb822b0ffe21c7d35837fc9f3d8",
-        timestamp: new Date("2018-06-16T05:59:49.500"),
-      },
-    })
+    expect(block.blockInfo.blockNumber).toEqual(20)
   })
 })

src/NodeosActionReader.ts

@@ -16,18 +16,17 @@ function wait(ms: number) {
  */
 export class NodeosActionReader extends AbstractActionReader {
   protected nodeosEndpoint: string
-  private log: Logger
+  protected log: Logger
 
   constructor(
     nodeosEndpoint: string = "http://localhost:8888",
     public startAtBlock: number = 1,
     protected onlyIrreversible: boolean = false,
     protected maxHistoryLength: number = 600,
-    protected requestInstance: any = request,
   ) {
     super(startAtBlock, onlyIrreversible, maxHistoryLength)
-    // Remove trailing slashes
-    this.nodeosEndpoint = nodeosEndpoint.replace(/\/+$/g, "")
+    this.nodeosEndpoint = nodeosEndpoint.replace(/\/+$/g, "") // Removes trailing slashes
+
     this.log = Logger.createLogger({ name: "demux" })
   }
 
@@ -36,54 +35,71 @@ export class NodeosActionReader extends AbstractActionReader {
    */
   public async getHeadBlockNumber(numRetries: number = 120, waitTimeMs: number = 250): Promise<number> {
     let numTries = 1
-    while (numTries < numRetries) {
+    while (numTries <= numRetries + 1) {
       try {
-        const blockInfo = await this.httpRequest("get", {
+        const blockInfo = await request.get({
           url: `${this.nodeosEndpoint}/v1/chain/get_info`,
           json: true,
         })
-        if (this.onlyIrreversible) {
-          return blockInfo.last_irreversible_block_num
-        }
         return blockInfo.head_block_num
       } catch (err) {
+        if (numTries - 1 === numRetries) {
+          throw err
+        }
         this.log.error("error getting head block number, retrying...")
       }
       numTries += 1
       await wait(waitTimeMs)
     }
 
-    throw Error("Retrieving head block number failed!")
+    throw Error("Unknown error getting head block number.")
+  }
+
+  public async getLastIrreversibleBlockNumber(numRetries: number = 120, waitTimeMs: number = 250): Promise<number> {
+    let numTries = 1
+    while (numTries <= numRetries + 1) {
+      try {
+        const blockInfo = await request.get({
+          url: `${this.nodeosEndpoint}/v1/chain/get_info`,
+          json: true,
+        })
+        return blockInfo.last_irreversible_block_num
+      } catch (err) {
+        if (numTries - 1 === numRetries) {
+          throw err
+        }
+        this.log.error("error getting last irreversible block number, retrying...")
+      }
+      numTries += 1
+      await wait(waitTimeMs)
+    }
+
+    throw Error("Unknown error getting last irreversible block number.")
   }
 
   /**
    * Returns a promise for a `NodeosBlock`.
    */
   public async getBlock(blockNumber: number, numRetries: number = 120, waitTimeMs: number = 250): Promise<NodeosBlock> {
     let numTries = 1
-    while (numTries < numRetries) {
+    while (numTries <= numRetries + 1) {
       try {
-        const rawBlock = await this.httpRequest("post", {
+        const rawBlock = await request.post({
           url: `${this.nodeosEndpoint}/v1/chain/get_block`,
           json: { block_num_or_id: blockNumber },
         })
         const block = new NodeosBlock(rawBlock)
         return block
       } catch (err) {
+        if (numTries - 1 === numRetries) {
+          throw err
+        }
         this.log.error("error retrieving block, retrying...")
       }
       numTries += 1
       await wait(waitTimeMs)
     }
 
-    throw Error("Retrieving block failed!")
-  }
-
-  protected async httpRequest(method: string, requestParams: any): Promise<any> {
-    if (method === "get") {
-      return await this.requestInstance.get(requestParams)
-    } else if (method === "post") {
-      return await this.requestInstance.post(requestParams)
-    }
+    throw Error("Unknown error getting block.")
   }
 }

yarn.lock

@@ -1825,10 +1825,12 @@ delegates@^1.0.0:
   resolved "https://registry.yarnpkg.com/delegates/-/delegates-1.0.0.tgz#84c6e159b81904fdca59a0ef44cd870d31250f9a"
   integrity sha1-hMbhWbgZBP3KWaDvRM2HDTElD5o=
 
-demux@^2.0.0:
-  version "2.0.0"
-  resolved "https://registry.yarnpkg.com/demux/-/demux-2.0.0.tgz#a5060bb1aac614a498ebe5ea181e9aa65e533ec1"
-  integrity sha512-zFXuTuvsBkcdHJn22Uc5T0wXnMxTv/auIDjNndhzXm2MLIdnXquE49G5i2davNifNCh+/kMnrWdzT9bKkjksRg==
+demux@^3.1.3:
+  version "3.1.3"
+  resolved "https://registry.yarnpkg.com/demux/-/demux-3.1.3.tgz#b72665ae469c309722d17e3ef5bb988018e712e2"
+  integrity sha512-hbillSK87rDhpN3A0ZRnw5vo2d1HnG1BbS/ud7W+dMsGSGbRKbg9gutpfPTqTQQrcqX4sNfLnWIpOTVfs808mw==
+  dependencies:
+    bunyan "^1.8.12"
 
 detect-indent@5.0.0:
   version "5.0.0"