const fs = require('fs')
const ip = require('ip')
const crypto = require('crypto')
const Cluster = require('../src/cluster')
const Connection = require('../src/network/connection')
const { createLogger, LEVELS: { NOTHING } } = require('../src/loggers')
const logFunctionConsole = require('../src/loggers/console')

const newLogger = () => createLogger({ level: NOTHING, logFunction: logFunctionConsole })
const getHost = () => process.env.HOST_IP || ip.address()
const secureRandom = (length = 10) => crypto.randomBytes(length).toString('hex')
const plainTextBrokers = (host = getHost()) => [`${host}:9092`, `${host}:9095`, `${host}:9098`]
const sslBrokers = (host = getHost()) => [`${host}:9093`, `${host}:9096`, `${host}:9099`]
const saslBrokers = (host = getHost()) => [`${host}:9094`, `${host}:9097`, `${host}:9100`]

const connectionOpts = () => ({
  clientId: `test-${secureRandom()}`,
  logger: newLogger(),
  host: getHost(),
  port: 9092,
})

const sslConnectionOpts = () =>
  Object.assign(connectionOpts(), {
    port: 9093,
    ssl: {
      servername: 'localhost',
      cert: fs.readFileSync('./testHelpers/certs/client_cert.pem', 'utf-8'),
      key: fs.readFileSync('./testHelpers/certs/client_key.pem', 'utf-8'),
      ca: [fs.readFileSync('./testHelpers/certs/ca_cert.pem', 'utf-8')],
    },
  })

const saslConnectionOpts = () =>
  Object.assign(sslConnectionOpts(), {
    port: 9094,
    sasl: {
      mechanism: 'plain',
      username: 'test',
      password: 'testtest',
    },
  })

const createConnection = (opts = {}) => new Connection(Object.assign(connectionOpts(), opts))
const createCluster = (opts = {}, brokers = plainTextBrokers()) =>
  new Cluster(Object.assign(connectionOpts(), opts, { brokers }))

const createModPartitioner = () => ({ partitionMetadata, message }) => {
  const numPartitions = partitionMetadata.length
  const key = parseInt(message.key.replace(/[^\d]/g, ''), 10)
  return ((key || 0) % 3) % numPartitions
}

const waitFor = (fn, { delay = 50 } = {}) => {
  return new Promise((resolve, reject) => {
    const check = () =>
      setTimeout(async () => {
        try {
          const result = await fn()
          result ? resolve(result) : check()
        } catch (e) {
          reject(e)
        }
      })
    check()
  })
}

const retryProtocol = (errorType, fn) =>
  waitFor(async () => {
    try {
      return await fn()
    } catch (e) {
      if (e.type !== errorType) {
        throw e
      }
      return false
    }
  })

const createTopic = (broker, topicName) =>
  retryProtocol('LEADER_NOT_AVAILABLE', async () => await broker.metadata([topicName]))

const waitForMessages = (buffer, { number = 1, delay = 50 } = {}) =>
  waitFor(() => (buffer.length >= number ? buffer.splice(0) : false), { delay })

module.exports = {
  secureRandom,
  connectionOpts,
  sslConnectionOpts,
  saslConnectionOpts,
  createConnection,
  createCluster,
  createModPartitioner,
  plainTextBrokers,
  sslBrokers,
  saslBrokers,
  newLogger,
  retryProtocol,
  createTopic,
  waitFor,
  waitForMessages,
}

@@ -2004,16 +2004,36 @@
 %7D) =%3E %7B%0A
+  let totalWait = 0%0A
   return
@@ -2091,16 +2091,43 @@
  = () =%3E
+ %7B%0A      totalWait += delay
 %0A      s
@@ -2194,24 +2194,33 @@
  = await fn(
+totalWait
 )%0A          
@@ -2317,20 +2317,42 @@
    %7D
-)%0A    check(
+, delay)%0A    %7D%0A    check(totalWait
 )%0A  

Fix test helper waitFor, make it actually wait