Home

Index

How to use the collect function from pull-stream

Find comprehensive JavaScript pull-stream.collect code examples handpicked from public code repositorys.

pull-stream.collect is a function in the pull-stream library that collects all the values of a stream into an array and passes it downstream as a single value.

operators.js

634
635
636
637
638
639
640
641
642
643
  fromDB(db),
  where(not(slowEqual('value.author', alice.id))),
  paginate(2),
  toPullStream()
),
pull.collect((err, pages) => {
  t.error(err, 'toPullStream got no error')
  t.equal(pages.length, 1, 'toPullStream got one page')
  const msgs = pages[0]
  t.equal(msgs.length, 1, 'page has one messages')

+ 3 other calls in file

index.js

GitHub: ssbc/ssb-db2

301
302
303
304
305
306
307
308
309
310

startMeasure(t, 'add 1000 box1 msgs')
pull(
  pull.values(contents),
  pull.asyncMap(sbot.db.publish),
  pull.collect((err, msgs) => {
    endMeasure(t, 'add 1000 box1 msgs')
    if (err) t.fail(err)

    sbot.db.onDrain('base', () => {

+ 5 other calls in file

How does pull-stream.collect work?

When you call pull-stream.collect with a pull-stream source, it creates a new through stream that collects all the values of the source stream and passes them downstream as a single value, an array containing all the collected values. Under the hood, pull-stream.collect creates a new through stream using pull.streams.through, which is a factory function for creating new through streams. This new through stream buffers all the incoming values from the source stream using an array, and when the source stream ends, it passes the buffered array downstream as a single value. When you pipe a source stream through the collect stream, it collects all the values of the source stream and passes them downstream as a single array value. This can be useful when you need to transform a stream of values into a single value, such as when you want to collect all the results of a database query or all the lines of a text file. Overall, pull-stream.collect simplifies working with pull-streams in JavaScript by providing an easy way to collect all the values of a stream into a single array value.

index.js

GitHub: ssbc/ssb-tribes

247
248
249
250
251
252
253
254
255
      paraMap(tribeGet, 4),
      opts.subtribes
        ? null
        : pull.filter(tribe => tribe.parentGroupId === undefined),
      pull.map(tribe => tribe.groupId),
      pull.collect(cb)
    )
  })
}

tumbling100.js

GitHub: pull-stream/pull-window

67
68
69
70
71
72
73
74
75
76
77
test('tumbling to 100', function (t) {


  pull(
    pull.count(100),
    groupTo100(),
    pull.collect(function (err, ary) {
      t.deepEqual(ary, expected)
      console.log(ary)
      t.end()
    })

Ai Example

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
const pull = require("pull-stream");
const collect = require("pull-stream/throughs/collect");

// Create a pull-stream source that emits some values
const source = pull.values([1, 2, 3, 4, 5]);

// Collect all the values of the source stream into an array
pull(
  source,
  collect((err, array) => {
    if (err) {
      console.error(err);
    } else {
      console.log(array);
    }
  })
);

In this example, we're using pull-stream.collect to collect all the values of a pull-stream source, which is created using pull.values. We then pipe the source stream through collect and pass a callback function to handle the collected array of values. When you run this code, you'll see that collect passes an array containing all the values of the source stream downstream as a single value. In this case, the output will be [1, 2, 3, 4, 5].

resolvers.js

GitHub: planetary-social/planetary-graphql

81
82
83
84
85
86
87
88
89
          .catch(cb)
      }, 5),
      pull.filter(Boolean),
      // TODO: This removes the profiles that came back as null, we might want to show something in place of that
      // e.g. someone who hasnt opted in to publicWebHosting
      pull.collect((err, res) => err ? reject(err) : resolve(res))
    )
  })
}

+ 20 other calls in file

index.js

GitHub: ssbc/ssb-tribes2

212
213
214
215
216
217
218
219
220
221
// prettier-ignore
if (err) return cb(clarify(err, 'Failed to get root metafeed when listing invites'))

pull(
  ssb.box2.listGroupIds(),
  pull.collect((err, groupIds) => {
    // prettier-ignore
    if (err) return cb(clarify(err, 'Failed to list group IDs when listing invites'))

    const source = pull(

+ 32 other calls in file

index.js

GitHub: ssbc/ssb-tribes2

193
194
195
196
197
198
199
200
201
202
// prettier-ignore
if (err) return cb(clarify(err, 'Failed to publish exclude msg'))

pull(
  listMembers(groupId),
  pull.collect((err, beforeMembers) => {
    // prettier-ignore
    if (err) return cb(clarify(err, "Couldn't get old member list when excluding members"))

    const remainingMembers = beforeMembers.filter(

+ 101 other calls in file

count-group-feeds.js

GitHub: ssbc/ssb-tribes2

7
8
9
10
11
12
13
14
15
16
pull(
  server.metafeeds.branchStream({ old: true, live: false }),
  pull.filter((branch) => branch.length === 4),
  pull.map((branch) => branch[3]),
  pull.filter((feed) => feed.recps),
  pull.collect((err, feeds) => {
    if (err) return cb(err)
    return cb(null, feeds.length)
  })
)

get-tangle-data.test.js

GitHub: ssbc/ssb-tribes2

42
43
44
45
46
47
48
49
50
51
  descending(),
  toPullStream()
),
pull.map((m) => fromMessageSigil(m.key)),
pull.take(1),
pull.collect((err, keys) => {
  t.error(err, 'no error')

  t.deepEqual(
    { root, previous },

+ 2 other calls in file

branch-stream.test.js

GitHub: ssbc/ssb-meta-feeds

13
14
15
16
17
18
19
20
21
22
const sbot = Testbot()

function testRead(t, sbot, cb) {
  pull(
    sbot.metafeeds.branchStream({ old: true, live: false }),
    pull.collect((err, branches) => {
      if (err) return cb(err)

      t.equal(branches.length, 7, '7 branches')
      const [

+ 4 other calls in file

ancestors.js

GitHub: elavoie/ssb-tokens

67
68
69
70
71
72
73
74
75
76
  console.log(output)
} else if (args.json) {
  pull(
    pull.values(keys),
    pull.asyncMap((msgId,cb) => ssb.get({ id: msgId, meta: true}, cb)),
    pull.collect(function (err, msgs) {
      var output = {}
      msgs.forEach(function (msg) {
        output[msg.key] = msg
      })

+ 78 other calls in file

add.js

GitHub: Paratii-Video/js-ipfs

61
62
63
64
65
66
67
68
69
addStream,
pull.map((file) => ({
  hash: file.hash,
  path: wrapWithDirectory ? file.path.substring(WRAPPER.length) : file.path
})),
pull.collect((err, added) => {
  if (err) {
    throw err
  }

+ 3 other calls in file

plugin.js

GitHub: achou11/ssb-storage-used

158
159
160
161
162
163
164
165
166
167
  gte: stringPrefix,
  lte: stringPrefix + '~',
  keys: true,
  values: true,
}),
pull.collect((err, chunk) => {
  prefix++
  cb(
    null,
    chunk

invite.js

GitHub: ssbc/ssb-invite-client

39
40
41
42
43
44
45
46
47
48
49
50
  .use(require('ssb-replication-scheduler'))
  // .use(require('ssb-invite'));
  .use(require('../lib'));


function all(stream, cb) {
  return pull(stream, pull.collect(cb));
}


function messagesByType(sbot, typeStr) {
  const {where, type, toPullStream} = sbot.db.operators;

test.js

GitHub: regular/flumeview-level-transform

42
43
44
45
46
47
48
49
50
51
sv.read({
  upto: true,
  keys: true,
  values: false
}),
pull.collect((err, data)=>{
  t.notOk(err, 'read() does not error')
  t.deepEqual(data, [
    {
      key: 1, seq: { foo: 'bar', bar: 'foo' }

send.js

GitHub: mkg20001/libp2p-tls

44
45
46
47
48
49
50
51
52
53
)

pull(
  remote,
  pull.take(100),
  pull.collect((err, chunks) => {
    if (err) throw err
    if (chunks.length !== 100) throw new Error('Did not receive enough chunks')
    deferred.resolve()
  })

muas_node.js

GitHub: tobed1995/cal-poly-muas-coin

101
102
103
104
105
106
107
108
109
110
if (err) {
    return
}
pull(
    conn,
    pull.collect(function (err, transaction) {
        if (transaction !== null, typeof transaction !== 'undefined' && transaction.length > 0) {
            let transObj = JSON.parse(transaction.join(''));
            //start calculating proof_of_work
            let p_o_w = self.proof_of_work(transObj).then(function (transObj) {

+ 2 other calls in file

infinite.js

GitHub: metapragma/valve

3
4
5
6
7
8
9
10
11
const pull = require('pull-stream')

const stream = pull(
  pull.infinite(),
  pull.take(10000),
  pull.collect((err, ary) => {
    console.log('completed')
  })
)

Other functions in pull-stream

Sorted by popularity

pull-stream.asyncMap is the most popular function in pull-stream (1458 examples)

Other popular packages:

Home

About