reduce: Throw TypeError if the channel contains no values and initialValue is not provided.

README.md

@@ -257,6 +257,21 @@ value has been shifted out or placed in the buffer.
   reserved value used to indicate a closed channel.
 
 #### reduce(callbackfn[, initialValue])
+
+`callbackfn` should be a function that takes two arguments (unlike `Array`'s
+version which takes four). `reduce` calls the callback, as a function, once for
+each value after the first value present in the channel.
+
+`callbackfn` is called with two arguments: the `previousValue` (value from the
+previous call to `callbackfn`) and the `currentValue`. The first time that
+callback is called, the `previousValue` and `currentValue` can be one of two
+values. If an `initialValue` was provided in the call to `reduce`, then
+`previousValue` will be equal to `initialValue` and `currentValue` will be equal
+to the first value in the channel. If no `initialValue` was provided, then
+`previousValue` will be equal to the first value in the channel and
+`currentValue` will be equal to the second. It is a `TypeError` if the channel
+contains no values and `initialValue` is not provided.
+
 #### shift() -> async
 
 Returns a promise that resolves when an value is received from the channel.

lib/index.js

@@ -182,14 +182,20 @@ const Channel = function (bufferLength = 0) {
 
       await readOnly.forEach((currentValue) => {
         if (previousValueDefined) {
-          previousValue = callbackfn(previousValue, currentValue, 0, readOnly)
+          previousValue = callbackfn(previousValue, currentValue)
         } else {
           previousValue = currentValue
           previousValueDefined = true
         }
       })
 
-      return previousValue
+      if (previousValueDefined) {
+        return previousValue
+      } else {
+        throw new TypeError(
+          `No values in channel and initialValue wasn't provided.`
+        )
+      }
     },
 
     shift: function () {

test/index.js

@@ -4,6 +4,20 @@ const assert = require('@nodeguy/assert')
 const Channel = require('../lib')
 const stream = require('stream')
 
+const assertRejects = async (callback, reason) => {
+  try {
+    await callback()
+  } catch (exception) {
+    if (reason) {
+      assert.deepEqual(exception, reason)
+    }
+
+    return
+  }
+
+  assert.fail(null, reason, `Missing expected rejection.`)
+}
+
 const toArray = async (channel) => {
   const array = []
 
@@ -147,35 +161,22 @@ describe(`Channel`, function () {
 
 describe(`Channel object`, function () {
   describe(`close`, function () {
-    it(`can't close an already closed channel`, function (done) {
+    it(`can't close an already closed channel`, function () {
       const channel = Channel()
       channel.close()
 
-      channel.close()
+      return assertRejects(async () => {
-        .then(() => {
+        await channel.close()
-          done(new Error())
+      }, new Error(`Can't close an already-closed channel.`))
-        })
-        .catch((reason) => {
-          assert.deepEqual(
-            reason,
-            new Error(`Can't close an already-closed channel.`)
-          )
-
-          done()
-        })
     })
 
     it(`can't push to a closed channel`, async function () {
       const channel = Channel()
       channel.close()
 
-      return (async () => {
+      return assertRejects(async () => {
         await channel.push(0)
-      })().then(() => {
+      }, new Error(`Can't push to closed channel.`))
-        assert(false)
-      }).catch((reason) => {
-        assert.deepEqual(reason, new Error(`Can't push to closed channel.`))
-      })
     })
 
     it(`returns 'undefined' immediately from shift`, async function () {
@@ -239,43 +240,31 @@ describe(`Channel object`, function () {
       it(`outside select`, function () {
         const channel = Channel()
 
-        return (async () => {
+        return assertRejects(
-          await channel.push(undefined)
+          async () => {
-        })().then(() => {
+            await channel.push(undefined)
-          assert(false)
+          },
-        }).catch((reason) => {
+          new TypeError(`Can't push 'undefined' to channel, use close instead.`)
-          assert.deepEqual(reason, new TypeError(
+        )
-            `Can't push 'undefined' to channel, use close instead.`
-          ))
-        })
       })
 
       it(`inside select`, function () {
         const channel = Channel()
 
-        return (async () => {
+        return assertRejects(async () => {
           await Channel.select(channel.push(undefined))
-        })().then(() => {
+        }, new TypeError(
-          assert(false)
+          `Can't push 'undefined' to channel, use close instead.`
-        }).catch((reason) => {
+        ))
-          assert.deepEqual(reason, new TypeError(
-            `Can't push 'undefined' to channel, use close instead.`
-          ))
-        })
       })
     })
 
     it(`disallows multiple values`, function () {
       const channel = Channel()
 
-      return (async () => {
+      return assertRejects(async () => {
         await channel.push(0, 1, 2)
-      })().then(() => {
+      }, new Error(`Can't push more than one value at a time.`))
-        assert(false)
-      }).catch((reason) => {
-        assert.deepEqual(reason, new Error(
-          `Can't push more than one value at a time.`))
-      })
     })
 
     it(`returns a frozen promise`, function () {
@@ -314,16 +303,29 @@ describe(`Channel object`, function () {
     })
   })
 
-  it(`reduce`, async function () {
+  describe(`reduce`, function () {
-    assert.equal(await Channel.of(0, 1, 2)
+    it(`callbackfn only`, async function () {
-      .reduce((previous, current) => previous + current),
+      assert.equal(await Channel.of(0, 1, 2)
-      3
+        .reduce(Math.max),
-    )
+        2
+      )
+    })
 
-    assert.equal(await Channel.of(0, 1, 2)
+    it(`initialValue`, async function () {
-      .reduce((previous, current) => previous + current, 10),
+      assert.equal(await Channel.of(0, 1, 2)
-      13
+        .reduce(Math.max, 10),
-    )
+        10
+      )
+    })
+
+    it(`no values without initialValue`, function () {
+      return assertRejects(
+        async () => {
+          await Channel.of().reduce(Math.max)
+        },
+        new TypeError(`No values in channel and initialValue wasn't provided.`)
+      )
+    })
   })
 
   describe(`shift`, function () {