replay-provider.ts

replay-provider.ts

import { getSyncProvider } from "@polkadot-api/json-rpc-provider-proxy";
import { PathLike } from "fs";
import { readFile } from "fs/promises";

interface QueueNode<T> {
  value: T
  next?: QueueNode<T>
}

class Queue<T> {
  private first?: QueueNode<T>
  private last?: QueueNode<T>

  constructor(...vals: T[]) {
    if (vals.length === 0) return
    vals.forEach((val) => this.push(val))
  }

  push(value: T) {
    const nextLast: QueueNode<T> = { value }
    if (this.last === undefined) {
      this.last = nextLast
      this.first = this.last
    } else {
      this.last.next = nextLast
      this.last = nextLast
    }
  }

  pop() {
    if (!this.first) throw new Error()
    const result = this.first.value

    this.first = this.first.next
    if (!this.first) {
      this.last = undefined
    }
    return result
  }

  isEmpty() {
    return !this.first
  }

  peek() {
    return this.first?.value
  }
}

export default (filePath: PathLike) => getSyncProvider(async () => {
  const contents = await readFile(filePath, { encoding: 'utf8' });
  const rawMessages = contents.split('\n').slice(0, -1)

  const messages = new Queue(...rawMessages.map(x => {
    const [, moment, direction, message] = x.match(/(\d*)-(..)-(.*)/)!
    return {
      moment: Number(moment),
      direction,
      message
    }
  }))

  const received = new Queue<string>()

  const start = async (listener: (msg: string) => void) => {
    do {
      if (messages.isEmpty()) {
        return
      }

      const current = messages.pop()

      if (current.direction === '<<') {
        listener(current.message)
          await Promise.resolve()
      } else {
        while (received.isEmpty())
          await new Promise(res => setTimeout(res, 0))

        const receivedMessage = received.pop()
        if (current.message !== receivedMessage) {
          console.error('Wrong message received')
          console.log(`received: ${receivedMessage}`)
          console.log(`expected: ${current.message}`)
          process.exit(1)
        }
      }
    } while(true)
  }

  return (listener) => {
    start(listener)

    return {
      send(msg: string) {
        received.push(msg)
      },
      disconnect() {}
    }
  }
})