add operator batch()

README.md

@@ -197,7 +197,7 @@ want to get results in batches, you should use **`toPullStream`**,
 **`paginate`**, and optionally `startFrom` and `descending`.
 
 - **toPullStream** creates a [pull-stream] source to stream the results
-- **paginate** configures the size of each page stream to the pull-stream source
+- **paginate** configures the size of each array sent to the pull-stream source
 - **startFrom** configures the beginning seq from where to start streaming
 - **descending** configures the pagination stream to order results
   from newest to oldest (otherwise the default order is oldest to
@@ -259,6 +259,36 @@ pull(
 )
 ```
 
+**Batching** with the operator `batch()` is similar to pagination in terms of
+performance, but the messages are delivered one-by-one to the final pull-stream,
+instead of as any array. Example:
+
+```js
+const pull = require('pull-stream')
+
+const source = query(
+  fromDB(db),
+  where(
+    and(
+      slowEqual('value.content.type', 'contact')
+      or(slowEqual('value.author', aliceId), slowEqual('value.author', bobId)),
+    ),
+  ),
+  batch(10), // Note `batch` instead of `paginate`
+  descending(),
+  toPullStream()
+)
+
+pull(
+  source,
+  // Note the below drain is `msg`, not `msgs` array:
+  pull.drain((msg) => {
+    console.log('next message:')
+    console.log(msg)
+  })
+)
+```
+
 #### async/await
 
 There are also operators that support getting the values using
@@ -372,6 +402,7 @@ const {
   offsets,
   count,
   paginate,
+  batch,
   startFrom,
   descending,
   asOffsets,

operators.js

@@ -323,6 +323,10 @@ function paginate(pageSize) {
   return (ops) => updateMeta(ops, 'pageSize', pageSize)
 }
 
+function batch(batchSize) {
+  return (ops) => updateMeta(ops, 'batchSize', batchSize)
+}
+
 function asOffsets() {
   return (ops) => updateMeta(ops, 'asOffsets', true)
 }
@@ -396,9 +400,9 @@ function toPullStream() {
     function paginateStream(ops) {
       let seq = meta.seq || 0
       let total = Infinity
-      const limit = meta.pageSize || 1
+      const limit = meta.pageSize || meta.batchSize || 20
       let shouldEnd = false
-      return function readable(end, cb) {
+      function readable(end, cb) {
         if (end) return cb(end)
         if (seq >= total || shouldEnd) return cb(true)
         if (meta.count) {
@@ -417,12 +421,18 @@ function toPullStream() {
               else {
                 total = answer.total
                 seq += limit
-                cb(null, !meta.pageSize ? answer.results[0] : answer.results)
+                cb(null, answer.results)
               }
             }
           )
         }
       }
+      if (meta.pageSize || meta.count) {
+        return readable
+      } else {
+        // Flatten the "pages" (arrays) into individual messages
+        return pull(readable, pull.map(pull.values), pull.flatten())
+      }
     }
 
     return pull(
@@ -480,6 +490,7 @@ module.exports = {
   count,
   startFrom,
   paginate,
+  batch,
   asOffsets,
   toCallback,
   toPullStream,

test/operators.js

@@ -26,6 +26,7 @@ const {
   offsets,
   fromDB,
   paginate,
+  batch,
   startFrom,
   live,
   count,
@@ -1257,6 +1258,7 @@ prepareAndRunTest('support live operations', dir, (t, db, raf) => {
       fromDB(db),
       where(slowEqual('value.content.type', 'post')),
       live({ old: true }),
+      batch(2),
       toPullStream(),
       pull.drain((msg) => {
         if (i++ == 0) {