<commit_before>/* eslint-env mocha */
/* eslint max-nested-callbacks: ['error', 8] */
'use strict'

const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)
const series = require('async/series')
const waterfall = require('async/waterfall')
const parallel = require('async/parallel')
const whilst = require('async/whilst')
const each = require('async/each')
const hat = require('hat')

function waitForPeers (ipfs, topic, peersToWait, callback) {
  const i = setInterval(() => {
    ipfs.pubsub.peers(topic, (err, peers) => {
      if (err) {
        return callback(err)
      }

      const missingPeers = peersToWait
        .map((e) => peers.indexOf(e) !== -1)
        .filter((e) => !e)

      if (missingPeers.length === 0) {
        clearInterval(i)
        callback()
      }
    })
  }, 500)
}

function spawnWithId (factory, callback) {
  waterfall([
    (cb) => factory.spawnNode(cb),
    (node, cb) => node.id((err, res) => {
      if (err) {
        return cb(err)
      }
      node.peerId = res
      cb(null, node)
    })
  ], callback)
}

function makeCheck (n, done) {
  let i = 0
  return (err) => {
    if (err) {
      return done(err)
    }

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

module.exports = (common) => {
  describe('.pubsub', function () {
    this.timeout(80 * 1000)

    const getTopic = () => 'pubsub-tests-' + hat()

    let ipfs1
    let ipfs2
    let ipfs3

    before(function (done) {
      // CI takes longer to instantiate the daemon, so we need to increase the
      // timeout for the before step
      this.timeout(100 * 1000)

      common.setup((err, factory) => {
        if (err) {
          return done(err)
        }

        series([
          (cb) => spawnWithId(factory, cb),
          (cb) => spawnWithId(factory, cb),
          (cb) => spawnWithId(factory, cb)
        ], (err, nodes) => {
          if (err) {
            return done(err)
          }

          ipfs1 = nodes[0]
          ipfs2 = nodes[1]
          ipfs3 = nodes[2]
          done()
        })
      })
    })

    after((done) => {
      common.teardown(done)
    })

    describe('single node', () => {
      describe('.publish', () => {
        it('errors on string messags', (done) => {
          const topic = getTopic()
          ipfs1.pubsub.publish(topic, 'hello friend', (err) => {
            expect(err).to.exist()
            done()
          })
        })

        it('message from buffer', (done) => {
          const topic = getTopic()
          ipfs1.pubsub.publish(topic, Buffer.from('hello friend'), done)
        })
      })

      describe('.subscribe', () => {
        it('to one topic', (done) => {
          const check = makeCheck(2, done)
          const topic = getTopic()

          const handler = (msg) => {
            expect(msg.data.toString()).to.equal('hi')
            expect(msg).to.have.property('seqno')
            expect(Buffer.isBuffer(msg.seqno)).to.eql(true)
            expect(msg).to.have.property('topicIDs').eql([topic])
            expect(msg).to.have.property('from', ipfs1.peerId.id)

            ipfs1.pubsub.unsubscribe(topic, handler)

            ipfs1.pubsub.ls((err, topics) => {
              expect(err).to.not.exist()
              expect(topics).to.be.empty()
              check()
            })
          }

          ipfs1.pubsub.subscribe(topic, handler, (err) => {
            expect(err).to.not.exist()
            ipfs1.pubsub.publish(topic, Buffer.from('hi'), check)
          })
        })

        it('attaches multiple event listeners', (done) => {
          const topic = getTopic()

          const check = makeCheck(3, done)
          const handler1 = (msg) => {
            expect(msg.data.toString()).to.eql('hello')

            ipfs1.pubsub.unsubscribe(topic, handler1)

            series([
              (cb) => ipfs1.pubsub.ls(cb),
              (cb) => {
                ipfs1.pubsub.unsubscribe(topic, handler2)
                cb()
              },
              (cb) => ipfs1.pubsub.ls(cb)
            ], (err, res) => {
              expect(err).to.not.exist()

              // Still subscribed as there is one listener left
              expect(res[0]).to.eql([topic])
              // Now all listeners are gone no subscription anymore
              expect(res[2]).to.eql([])
              check()
            })
          }

          const handler2 = (msg) => {
            expect(msg.data.toString()).to.eql('hello')<commit_msg>test(pubsub): subscribe with a Promise (#181)<commit_after>/* eslint-env mocha */
/* eslint max-nested-callbacks: ['error', 8] */
'use strict'

const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)
const series = require('async/series')
const waterfall = require('async/waterfall')
const parallel = require('async/parallel')
const whilst = require('async/whilst')
const each = require('async/each')
const hat = require('hat')

function waitForPeers (ipfs, topic, peersToWait, callback) {
  const i = setInterval(() => {
    ipfs.pubsub.peers(topic, (err, peers) => {
      if (err) {
        return callback(err)
      }

      const missingPeers = peersToWait
        .map((e) => peers.indexOf(e) !== -1)
        .filter((e) => !e)

      if (missingPeers.length === 0) {
        clearInterval(i)
        callback()
      }
    })
  }, 500)
}

function spawnWithId (factory, callback) {
  waterfall([
    (cb) => factory.spawnNode(cb),
    (node, cb) => node.id((err, res) => {
      if (err) {
        return cb(err)
      }
      node.peerId = res
      cb(null, node)
    })
  ], callback)
}

function makeCheck (n, done) {
  let i = 0
  return (err) => {
    if (err) {
      return done(err)
    }

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

module.exports = (common) => {
  describe('.pubsub', function () {
    this.timeout(80 * 1000)

    const getTopic = () => 'pubsub-tests-' + hat()

    let ipfs1
    let ipfs2
    let ipfs3

    before(function (done) {
      // CI takes longer to instantiate the daemon, so we need to increase the
      // timeout for the before step
      this.timeout(100 * 1000)

      common.setup((err, factory) => {
        if (err) {
          return done(err)
        }

        series([
          (cb) => spawnWithId(factory, cb),
          (cb) => spawnWithId(factory, cb),
          (cb) => spawnWithId(factory, cb)
        ], (err, nodes) => {
          if (err) {
            return done(err)
          }

          ipfs1 = nodes[0]
          ipfs2 = nodes[1]
          ipfs3 = nodes[2]
          done()
        })
      })
    })

    after((done) => {
      common.teardown(done)
    })

    describe('single node', () => {
      describe('.publish', () => {
        it('errors on string messags', (done) => {
          const topic = getTopic()
          ipfs1.pubsub.publish(topic, 'hello friend', (err) => {
            expect(err).to.exist()
            done()
          })
        })

        it('message from buffer', (done) => {
          const topic = getTopic()
          ipfs1.pubsub.publish(topic, Buffer.from('hello friend'), done)
        })
      })

      describe('.subscribe', () => {
        it('to one topic', (done) => {
          const check = makeCheck(2, done)
          const topic = getTopic()

          const handler = (msg) => {
            expect(msg.data.toString()).to.equal('hi')
            expect(msg).to.have.property('seqno')
            expect(Buffer.isBuffer(msg.seqno)).to.eql(true)
            expect(msg).to.have.property('topicIDs').eql([topic])
            expect(msg).to.have.property('from', ipfs1.peerId.id)

            ipfs1.pubsub.unsubscribe(topic, handler)

            ipfs1.pubsub.ls((err, topics) => {
              expect(err).to.not.exist()
              expect(topics).to.be.empty()
              check()
            })
          }

          ipfs1.pubsub.subscribe(topic, handler, (err) => {
            expect(err).to.not.exist()
            ipfs1.pubsub.publish(topic, Buffer.from('hi'), check)
          })
        })

        it('to one topic with Promise', (done) => {
          const check = makeCheck(2, done)
          const topic = getTopic()

          const handler = (msg) => {
            expect(msg.data.toString()).to.equal('hi')
            expect(msg).to.have.property('seqno')
            expect(Buffer.isBuffer(msg.seqno)).to.eql(true)
            expect(msg).to.have.property('topicIDs').eql([topic])
            expect(msg).to.have.property('from', ipfs1.peerId.id)

            ipfs1.pubsub.unsubscribe(topic, handler)

            ipfs1.pubsub.ls((err, topics) => {
              expect(err).to.not.exist()
              expect(topics).to.be.empty()
              check()
            })
          }

          ipfs1.pubsub
            .subscribe(topic, handler)
            .then(() => ipfs1.pubsub.publish(topic, Buffer.from('hi'), check))
            .catch((err) => expect(err).to.not.exist())
        })

        it('to one topic with options and Promise', (done) => {
          const check = makeCheck(2, done)
          const topic = getTopic()

          const handler = (msg) => {
            expect(msg.data.toString()).to.equal('hi')
            expect(msg).to.have.property('seqno')
            expect(Buffer.isBuffer(msg.seqno)).to.eql(true)
            expect(msg).to.have.property('topicIDs').eql([topic])
            expect(msg).to.have.property('from', ipfs1.peerId.id)

            ipfs1.pubsub.unsubscribe(topic, handler)

            ipfs1.pubsub.ls((err, topics) => {
              expect(err).to.not.exist()
              expect(topics).to.be.empty()
              check()
            })
          }

          ipfs1.pubsub
            .subscribe(topic, {}, handler)
            .then(() => ipfs1.pubsub.publish(topic, Buffer.from('hi'), check))
            .catch((err) => expect(err).to.not.exist())
        })

        it('attaches multiple event listeners', (done) => {
          const topic = getTopic()

          const check = makeCheck(3, done)
          const handler1 = (msg) => {
            expect(msg.data.toString()).to.eql('hello')

            ipfs1.pubsub.unsubscribe(topic, handler1)

            series([
              (cb) => ipfs1.pubsub.ls(cb),
              (cb) => {
                ipfs1.pubsub.unsubscribe(topic, handler2)
                cb()
              },
              (cb) => ipfs1.pubsub.ls(cb)
            ], (err, res) => {
              expect(err).to.not.exist()

              // Still subscribed as there is one listener left
              expect(res[0]).to.eql([topic])
              // Now all listeners are gone no subscription anymore
              expect(res[2]).to.eql([])
              check()
            })
          }

          const handler2 = (msg) => {
            expect(msg.data.toString()).to.eql('hello')
{'fraction_non_alphanumeric': 0.1591952967113724, 'fraction_numerical': 0.010196582766856513, 'mean_word_length': 1.7610956124778088, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3459363', 'n_tokens_mistral': 3433, 'n_tokens_neox': 3327, 'n_words': 660}