Add block event listener config

Signed-off-by: James Taylor <jamest@uk.ibm.com>

asset-transfer-basic/rest-api-typescript/src/config.spec.ts

@@ -152,6 +152,28 @@ describe('Config values', () => {
     });
   });
 
+  describe('blockListenerOrg', () => {
+    it('defaults to "Org1"', () => {
+      const config = require('./config');
+      expect(config.blockListenerOrg).toBe('Org1');
+    });
+
+    it('can be configured using the "HLF_BLOCK_LISTENER_ORG" environment variable', () => {
+      process.env.HLF_BLOCK_LISTENER_ORG = 'Org2';
+      const config = require('./config');
+      expect(config.blockListenerOrg).toBe('Org2');
+    });
+
+    it('throws an error when the "HLF_BLOCK_LISTENER_ORG" environment variable has an invalid value', () => {
+      process.env.HLF_BLOCK_LISTENER_ORG = 'Org3';
+      expect(() => {
+        require('./config');
+      }).toThrow(
+        'env-var: "HLF_BLOCK_LISTENER_ORG" should be one of [Org1, Org2]'
+      );
+    });
+  });
+
   describe('channelName', () => {
     it('defaults to "mychannel"', () => {
       const config = require('./config');

asset-transfer-basic/rest-api-typescript/src/config.ts

@@ -4,6 +4,9 @@
 
 import * as env from 'env-var';
 
+export const ORG1 = 'Org1';
+export const ORG2 = 'Org2';
+
 /*
  * Log level for the REST server
  */
@@ -55,8 +58,8 @@ export const asLocalhost = env
  */
 export const mspIdOrg1 = env
   .get('HLF_MSP_ID_ORG1')
-  .default('Org1MSP')
-  .example('Org1MSP')
+  .default(`${ORG1}MSP`)
+  .example(`${ORG1}MSP`)
   .asString();
 
 /*
@@ -64,10 +67,18 @@ export const mspIdOrg1 = env
  */
 export const mspIdOrg2 = env
   .get('HLF_MSP_ID_ORG2')
-  .default('Org2MSP')
-  .example('Org2MSP')
+  .default(`${ORG2}MSP`)
+  .example(`${ORG2}MSP`)
   .asString();
 
+/*
+ * The block listener org
+ */
+export const blockListenerOrg = env
+  .get('HLF_BLOCK_LISTENER_ORG')
+  .default(ORG1)
+  .asEnum([ORG1, ORG2]);
+
 /*
  * Name of the channel which the basic asset sample chaincode has been installed on
  */

asset-transfer-basic/rest-api-typescript/src/fabric.spec.ts

@@ -11,6 +11,7 @@ import {
   submitTransaction,
   getBlockHeight,
   startRetryLoop,
+  blockEventHandler,
 } from './fabric';
 import * as config from './config';
 
@@ -22,11 +23,13 @@ import {
 } from './errors';
 
 import {
+  BlockEvent,
   Contract,
   Gateway,
   GatewayOptions,
   Network,
   Transaction,
+  TransactionEvent,
   Wallet,
 } from 'fabric-network';
 
@@ -40,6 +43,36 @@ jest.mock('./config');
 jest.mock('ioredis', () => require('ioredis-mock/jest'));
 
 describe('Fabric', () => {
+  const mockTransactionId =
+    '0ae62c01e4c4b112c3f3954a2f11243da76778e46df9ad2783bcbafc79652b95';
+  const mockKey = `txn:${mockTransactionId}`;
+  const mockMspId = 'Org1MSP';
+  const mockState = Buffer.from(
+    `{"name":"CreateAsset","nonce":"damqinq8nrI4n4qY8lFVsZw7RwG2ufrv","transactionId":${mockTransactionId}`
+  );
+  const mockArgs = '["test111","red",400,"Jean",101]';
+  const mockTimestamp = 1628078044362;
+
+  const addMockTransationDetails = async (redis: Redis) => {
+    await redis
+      .multi()
+      .hset(
+        mockKey,
+        'mspId',
+        mockMspId,
+        'state',
+        mockState,
+        'args',
+        mockArgs,
+        'timestamp',
+        mockTimestamp,
+        'retries',
+        '0'
+      )
+      .zadd('index:txn:timestamp', mockTimestamp, mockTransactionId)
+      .exec();
+  };
+
   describe('createWallet', () => {
     it('creates a wallet containing identities for both orgs', async () => {
       const wallet = await createWallet();
@@ -110,41 +143,11 @@ describe('Fabric', () => {
     let mockContract: MockProxy<Contract>;
     let mockContracts: Map<string, Contract>;
 
-    const mockTransactionId =
-      '0ae62c01e4c4b112c3f3954a2f11243da76778e46df9ad2783bcbafc79652b95';
-    const mockKey = `txn:${mockTransactionId}`;
-    const mockMspId = 'Org1MSP';
-    const mockState = Buffer.from(
-      `{"name":"CreateAsset","nonce":"damqinq8nrI4n4qY8lFVsZw7RwG2ufrv","transactionId":${mockTransactionId}`
-    );
-    const mockArgs = '["test111","red",400,"Jean",101]';
-    const mockTimestamp = 1628078044362;
-
     const flushPromises = () => {
       jest.useRealTimers();
       return new Promise((resolve) => setImmediate(resolve));
     };
 
-    const addMockTransationDetails = async (redis: Redis) => {
-      await redis
-        .multi()
-        .hset(
-          mockKey,
-          'mspId',
-          mockMspId,
-          'state',
-          mockState,
-          'args',
-          mockArgs,
-          'timestamp',
-          mockTimestamp,
-          'retries',
-          '0'
-        )
-        .zadd('index:txn:timestamp', mockTimestamp, mockTransactionId)
-        .exec();
-    };
-
     beforeEach(() => {
       const redisOptions = {
         port: config.redisPort,
@@ -485,6 +488,63 @@ describe('Fabric', () => {
     });
   });
 
+  describe('blockEventHandler', () => {
+    let redis: Redis;
+    let mockIsValidGetter: jest.Mock<boolean, []>;
+    let mockTransactionIdGetter: jest.Mock<string, []>;
+    let mockTransactionEvent: MockProxy<TransactionEvent>;
+    let mockBlockEvent: MockProxy<BlockEvent>;
+
+    beforeEach(async () => {
+      const redisOptions = {
+        port: config.redisPort,
+        host: config.redisHost,
+        username: config.redisUsername,
+        password: config.redisPassword,
+      };
+
+      redis = new IORedis(redisOptions) as unknown as Redis;
+      addMockTransationDetails(redis);
+
+      const baseMock = {};
+      mockTransactionEvent = mock<TransactionEvent>(baseMock);
+      mockIsValidGetter = jest.fn<boolean, []>();
+      Object.defineProperty(baseMock, 'isValid', { get: mockIsValidGetter });
+      mockTransactionIdGetter = jest.fn<string, []>();
+      Object.defineProperty(baseMock, 'transactionId', {
+        get: mockTransactionIdGetter,
+      });
+
+      mockBlockEvent = mock<BlockEvent>();
+      mockBlockEvent.getTransactionEvents.mockReturnValue([
+        mockTransactionEvent,
+      ]);
+    });
+
+    it('clears saved details for valid transactions', async () => {
+      const blockListener = blockEventHandler(redis);
+      mockIsValidGetter.mockReturnValue(true);
+      mockTransactionIdGetter.mockReturnValue(mockTransactionId);
+
+      await blockListener(mockBlockEvent);
+
+      const index = await redis.zrange('index:txn:timestamp', 0, -1);
+      expect(index).toStrictEqual([]);
+    });
+
+    it('does not clear saved details for invalid transactions', async () => {
+      const blockListener = blockEventHandler(redis);
+      mockIsValidGetter.mockReturnValue(false);
+
+      await blockListener(mockBlockEvent);
+
+      const index = await redis.zrange('index:txn:timestamp', 0, -1);
+      expect(index).toStrictEqual([
+        '0ae62c01e4c4b112c3f3954a2f11243da76778e46df9ad2783bcbafc79652b95',
+      ]);
+    });
+  });
+
   describe('getBlockHeight', () => {
     it('gets the current block height', async () => {
       const mockBlockchainInfoProto =

asset-transfer-basic/rest-api-typescript/src/fabric.ts

@@ -344,23 +344,35 @@ const isDuplicateTransactionError = (error: {
   return false;
 };
 
+/*
+ * Block event listener to handle successful transactions
+ *
+ * Transaction details are saved before being submitted so that
+ * they can be retried, and this listener deletes those transaction
+ * details for any successful transactions
+ *
+ * Transactions can be submitted using one of two identities
+ * however one one of those identities is used to listen for
+ * block events
+ */
 export const blockEventHandler = (redis: Redis): BlockListener => {
-  const blockListner = async (event: BlockEvent) => {
-    logger.debug('Block event received ');
-    const transEvents: Array<TransactionEvent> = event.getTransactionEvents();
+  const blockListener = async (event: BlockEvent) => {
+    logger.debug(
+      { blockNumber: event.blockNumber.toString() },
+      'Block event received'
+    );
+    const transactionEvents: Array<TransactionEvent> =
+      event.getTransactionEvents();
 
-    for (const transEvent of transEvents) {
-      if (transEvent && transEvent.isValid) {
-        logger.debug(
-          'Remove transation with txnId %s',
-          transEvent.transactionId
-        );
-        await clearTransactionDetails(redis, transEvent.transactionId);
+    for (const event of transactionEvents) {
+      if (event && event.isValid) {
+        logger.debug('Remove transation with txnId %s', event.transactionId);
+        await clearTransactionDetails(redis, event.transactionId);
       }
     }
   };
 
-  return blockListner;
+  return blockListener;
 };
 
 export const getBlockHeight = async (

asset-transfer-basic/rest-api-typescript/src/index.ts

@@ -2,22 +2,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { Network } from 'fabric-network';
-import { Redis } from 'ioredis';
 import * as config from './config';
-import { blockEventHandler } from './fabric';
 import { logger } from './logger';
 import { createServer } from './server';
 
 async function main() {
   const app = await createServer();
 
-  // TODO block listener currently only handles a single org!!!
-  // TODO should it be initialised here?
-  const redis = app.get('redis') as Redis;
-  const network = app.get('networkOrg1') as Network;
-  await network.addBlockListener(blockEventHandler(redis));
-
   app.listen(config.port, () => {
     logger.info('Express server started on port: %d', config.port);
   });

asset-transfer-basic/rest-api-typescript/src/server.ts

@@ -18,6 +18,7 @@ import {
   createGateway,
   createWallet,
   startRetryLoop,
+  blockEventHandler,
 } from './fabric';
 import { redis } from './redis';
 import * as config from './config';
@@ -81,9 +82,6 @@ export const createServer = async (): Promise<Application> => {
   const contractsOrg1 = await getContracts(networkOrg1);
   app.set(config.mspIdOrg1, contractsOrg1);
 
-  // TODO used for block listener, which needs fixing!
-  app.set('networkOrg1', networkOrg1);
-
   const gatewayOrg2 = await createGateway(
     config.connectionProfileOrg2,
     config.mspIdOrg2,
@@ -100,6 +98,13 @@ export const createServer = async (): Promise<Application> => {
 
   app.set('redis', redis);
 
+  logger.debug('Adding block listener to %s network', config.blockListenerOrg);
+  if (config.blockListenerOrg === config.ORG1) {
+    await networkOrg1.addBlockListener(blockEventHandler(redis));
+  } else {
+    await networkOrg2.addBlockListener(blockEventHandler(redis));
+  }
+
   app.use('/', healthRouter);
   app.use('/api/assets', authenticateApiKey, assetsRouter);
   app.use('/api/transactions', authenticateApiKey, transactionsRouter);