How to use the streaming-iterables.collect function in streaming-iterables

await Promise.all([
          ipfs1.pubsub.subscribe(topic, sub),
          ipfs2.pubsub.subscribe(topic, () => {})
        ])

        await waitForPeers(ipfs1, topic, [ipfs2.peerId.id], 30000)

        const startTime = new Date().getTime()

        for (let i = 0; i < count; i++) {
          const msgData = Buffer.from(msgBase + i)
          await ipfs2.pubsub.publish(topic, msgData)
        }

        const msgs = await collect(msgStream)
        const duration = new Date().getTime() - startTime
        const opsPerSec = Math.floor(count / (duration / 1000))

        // eslint-disable-next-line
        console.log(`Send/Receive 100 messages took: ${duration} ms, ${opsPerSec} ops / s`)

        msgs.forEach(msg => {
          expect(msg.from).to.eql(ipfs2.peerId.id)
          expect(msg.data.toString().startsWith(msgBase)).to.be.true()
        })
      })
    })

const handler2 = msg => {
          msgStream2.push(msg)
          msgStream2.end()
        }

        await Promise.all([
          ipfs1.pubsub.subscribe(topic, handler1),
          ipfs1.pubsub.subscribe(topic, handler2)
        ])

        await ipfs1.pubsub.publish(topic, Buffer.from('hello'))

        const [handler1Msg] = await collect(msgStream1)
        expect(handler1Msg.data.toString()).to.eql('hello')

        const [handler2Msg] = await collect(msgStream2)
        expect(handler2Msg.data.toString()).to.eql('hello')

        await ipfs1.pubsub.unsubscribe(topic, handler1)
        await delay(100)

        // Still subscribed as there is one listener left
        expect(await ipfs1.pubsub.ls()).to.eql([topic])

        await ipfs1.pubsub.unsubscribe(topic, handler2)
        await delay(100)

        // Now all listeners are gone no subscription anymore
        expect(await ipfs1.pubsub.ls()).to.eql([])
      })

}
        const sub2 = msg => {
          msgStream2.push(msg)
          msgStream2.end()
        }

        await Promise.all([
          ipfs1.pubsub.subscribe(topic, sub1),
          ipfs2.pubsub.subscribe(topic, sub2)
        ])

        await waitForPeers(ipfs2, topic, [ipfs1.peerId.id], 30000)

        await ipfs2.pubsub.publish(topic, buffer)

        const [sub1Msg] = await collect(msgStream1)
        expect(sub1Msg.data.toString('hex')).to.be.eql(expectedHex)
        expect(sub1Msg.from).to.eql(ipfs2.peerId.id)

        const [sub2Msg] = await collect(msgStream2)
        expect(sub2Msg.data.toString('hex')).to.be.eql(expectedHex)
        expect(sub2Msg.from).to.eql(ipfs2.peerId.id)
      })