generic balanced reducer

src/importer/builder/balanced/reducer.js

@@ -0,0 +1,87 @@
+'use strict'
+
+const pull = require('pull-stream')
+const pullWrite = require('pull-write')
+const pushable = require('pull-pushable')
+const batch = require('pull-batch')
+const through = require('pull-through')
+
+module.exports = function balancedReduceToRoot(specialization, options) {
+
+  const deliverRoot = pushable()
+  const refeed = pushable()
+
+  const sink = pullWrite(
+    function (items, cb) { // write
+      refeed.push(items)
+      // items.forEach(item => refeed.push(item))
+      cb()
+    },
+    null, // reduce
+    1, // max
+    function (err) {
+      refeed.end()
+    }
+  )
+
+
+  let root
+
+  pull(
+    refeed,
+    batch(options.maxChildrenPerNode),
+    through(
+      function(items) {
+        if (items.length > 1) {
+          specialization.reduce(items, function(err, parent) {
+            if (err) {
+              this.queue(err)
+              return
+            }
+
+            root = parent
+            refeed.push(root)
+          })
+        } else {
+          this.queue(items[0])
+         }
+      },
+      function (end) {
+        if (root) {
+          this.queue(root)
+        }
+        this.queue(null)
+      }
+    ),
+    pull.take(1),
+    pull.drain(function (root) {
+      deliverRoot.push(root)
+      deliverRoot.end()
+    })
+  )
+
+  return {
+    source: deliverRoot,
+    sink: sink
+  }
+}
+
+// const specialization = {
+//   reduce: function (items, callback) {
+//     callback(null, {
+//       children: items
+//     })
+//   }
+// }
+
+// const options = {
+//   maxChildrenPerNode: 5
+// }
+
+// pull(
+//   pull.values([1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ,11, 12, 13, 14]),
+//   balancedReduceToRoot(specialization, options),
+//   pull.drain((root) => {
+//     console.log('root is: %s', JSON.stringify(root, undefined, '  '))
+//   })
+// )