GRPC listener

Config interface

export interface ConfigRpc {
  addresses: string[] // grpc node's addresses
  crtFile?: any // .pem file for tls (if enabled)
  logger: ILogger // logger
  auth: ApiKeyAuth | ServiceTokenAuth // auth
  connectionId?: string // GRPC connectionId using in blockchain Node
  getLastBlocksSignature?: () => Promise<string> // function to get last block signature, if undefined => parse blockchain from begining
  asyncGrpc?: boolean // pause stream to avoid cache overflow, wait until 
  logTimers?: boolean // log different timers
  txLifetime?: number // cash tx lifetime
  filters?: {
    tx_types: number[] // tx types filter
  }
}

export interface ApiKeyAuth {
  nodeApiKey: string
}

export interface ServiceTokenAuth {
  serviceToken: string,
  authServiceAddress: string
}

Required functions

Need to implement such methods in your system

type RollbackLastBlock = () => Promise<any> // rollback last block
type RollbackToBlockSignature = (signature: string) => Promise<any> // rollback to specified block 
type ReceiveTxs = (block: NodeBlock, txs: ParsedIncomingGrpcTxType[]) => Promise<any> // take your transactions and block 

// optional
type ReceiveNewGrpcCall = (call: grpc.ClientReadableStream<any>) => any // take stream object if you need
type ReceiveCriticalError = (err: Error) => any // handle critical error (default - kill node.js process)

type OnHistorySynced = () => void // trigger when receive all HistoryBlocks

Example implementation

import { ParsedIncomingGrpcTxType } from '@wavesenterprise/js-sdk'
import { GrpcListener, ConfigRpc } from '@wavesenterprise/grpc-listener'

export class RPCSyncService {
  config: ConfigRpc
  listener: GrpcListener
  call: grpc.ClientReadableStream<any>

  constructor (readonly addresses: string[]) {
    const crt = GRPC_CERT_FILE_PATH && fs.readFileSync(GRPC_CERT_FILE_PATH)
    this.config = {
      addresses: this.addresses,
      crtFile: crt,
      logger,
      auth: NODE_API_KEY
        ? {nodeApiKey: NODE_API_KEY}
        : {serviceToken: SERVICE_TOKEN, authServiceAddress: AUTH_SERVICE_ADDRESS},
      asyncGrpc: ASYNC_GRPC,
      logTimers: true,
      txLifetime: TX_LIFETIME,
      getLastBlocksSignature: this.getLastBlocksSignature
    }
  }

  async start() {
    this.listener = new GrpcListener(this.config)
    await this.listener.listen(
      rollbackLastBlock: this.persistService.rollbackLastBlock,
      rollbackToBlockSignature: this.persistService.rollbackToBlockSignature,
      receiveTxs: this.receiveTxs,
      receiveCriticalError: this.receiveError,
      receiveNewGrpcCall: this.receiveNewGrpcCall,
      onHistorySynced: this.onHistorySynced
    )
  }

  onHistorySynced = () => {
    // blockchain parsed
  }

  receiveTxs = async (block: NodeBlock, txs: ParsedIncomingGrpcTxType[]) => {
    // take your txs here
  }

  receiveNewGrpcCall = (call: grpc.ClientReadableStream<any>) => {
    // take if you need
    this.call = call
  }

  getLastBlocksSignature = async () => {
    // TODO
    return Promise.resolve('test')
  }

  rollbackLastBlock = async () => {
    // TODO
    return Promise.resolve()
  }

  rollbackToBlockSignature = async (signature: string) => {
    // TODO
    return Promise.resolve()
  }

  receiveError = async (err: Error) => {
    console.trace(err);
    process.exit(1)
  }
}