pull-plex/profile.js at master · dryajov/pull-plex · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
'use strict'

const pair = require('pull-pair/duplex')
const pull = require('pull-stream')
const generate = require('pull-generate')
const each = require('async/each')
const eachLimit = require('async/eachLimit')
const setImmediate = require('async/setImmediate')

const Plex = require('./src')

const spawn = (nStreams, nMsg, done, limit) => {
  const p = pair()

  const check = marker(2 * nStreams, done)

  const msg = 'simple msg'

  const listener = new Plex(false)
  const dialer = new Plex(true)

  pull(dialer, p[0], dialer)
  pull(listener, p[1], listener)

  listener.on('stream', (stream) => {
    pull(
      stream,
      pull.onEnd((err) => {
        if (err) { return done(err) }
        check()
        pull(pull.empty(), stream)
      })
    )
  })

  const numbers = []
  for (let i = 0; i < nStreams; i++) {
    numbers.push(i)
  }

  const spawnStream = (n, cb) => {
    const stream = dialer.createStream()
    pull(
      generate(0, (s, cb) => {
        setImmediate(() => {
          cb(s === nMsg ? true : null, msg, s + 1)
        })
      }),
      stream,
      pull.collect((err) => {
        if (err) { return done(err) }
        check()
        cb()
      })
    )
  }

  if (limit) {
    eachLimit(numbers, limit, spawnStream, () => {})
  } else {
    each(numbers, spawnStream, () => {})
  }
}

function marker (n, done) {
  let i = 0
  return (err) => {
    i++

    // console.log(`${i} out of ${n} interactions`)
    if (err) {
      console.error('Failed after %s iterations', i)
      return done(err)
    }

    if (i === n) {
      done()
    }
  }
}


spawn(1000, 1000, (err) => {
  if (err) {
    throw err
  }
  console.log('Done')
  process.exit(0)
}, 50000)