Add c8 for code coverage reporting fail on less than 95% coverage

.github/workflows/test.yml

@@ -20,3 +20,7 @@ jobs:
         npm test
       env:
         CI: true
+    - name: Check Coverage Levels
+      if: ${{ matrix.node-version != '10.x' }}
+      run: |
+        npm run check-coverage

.gitignore

@@ -1,3 +1,4 @@
-node_modules
+coverage
 dist
 dist-ts
+node_modules

.npmignore

@@ -5,6 +5,7 @@
 api-extractor.json
 benchmark
 budle-types.js
+coverage
 dist-ts
 lib
 package-lock.json

lib/batch-test.ts

@@ -21,26 +21,18 @@ function* numbers(max: number) {
 describe('batch', () => {
   it('batches async iterators', async () => {
     const batches: number[][] = []
-    for await (const numberBatch of batch(5, asyncNumbers(10))) {
-      assert.equal(numberBatch.length, 5)
+    for await (const numberBatch of batch(5, asyncNumbers(11))) {
       batches.push(numberBatch)
     }
-    assert.deepEqual(batches, [
-      [1, 2, 3, 4, 5],
-      [6, 7, 8, 9, 10],
-    ])
+    assert.deepEqual(batches, [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10], [11]])
   })
 
   it('batches sync iterators', async () => {
     const batches: number[][] = []
-    for await (const numberBatch of batch(5, numbers(10))) {
-      assert.equal(numberBatch.length, 5)
+    for await (const numberBatch of batch(5, numbers(11))) {
       batches.push(numberBatch)
     }
-    assert.deepEqual(batches, [
-      [1, 2, 3, 4, 5],
-      [6, 7, 8, 9, 10],
-    ])
+    assert.deepEqual(batches, [[1, 2, 3, 4, 5], [6, 7, 8, 9, 10], [11]])
   })
 
   it('is curryable', async () => {

lib/flat-map-test.ts

@@ -27,4 +27,12 @@ describe('flatmap', () => {
     }
     assert.deepEqual(numbers, [1, 2, 3, 4, 5, 6, 7, 8])
   })
+  it('curries', async () => {
+    const numbers: number[] = []
+    const mapNothing = flatMap(i => i)
+    for await (const num of mapNothing([1, 2, [3, 4, 5, 6], 7, [8]])) {
+      numbers.push(num as any)
+    }
+    assert.deepEqual(numbers, [1, 2, 3, 4, 5, 6, 7, 8])
+  })
 })

lib/flat-map.ts

@@ -8,7 +8,7 @@ export function flatMap<T, B>(func: (data: T) => FlatMapValue<B>): (iterable: An
 export function flatMap<T, B>(func: (data: T) => FlatMapValue<B>, iterable: AnyIterable<T>): AsyncGenerator<B>
 export function flatMap<T, B>(func: (data: T) => FlatMapValue<B>, iterable?: AnyIterable<T>) {
   if (iterable === undefined) {
-    return (curriedIterable: AnyIterable<T>) => flatMap(func, curriedIterable)
+    return curriedIterable => flatMap(func, curriedIterable)
   }
   return filter<B>(i => i !== undefined && i !== null, flatten(map<any, any>(func, iterable)))
 }

lib/from-stream-test.ts

@@ -2,6 +2,17 @@ import { assert } from 'chai'
 import { fromStream } from '.'
 import { PassThrough } from 'stream'
 
+const once = func => {
+  let called = false
+  return (...args) => {
+    if (called) {
+      return
+    }
+    called = true
+    return func(...args)
+  }
+}
+
 describe('fromStream', () => {
   it('takes a stream and returns an async iterable', async () => {
     const stream = new PassThrough({ objectMode: true })
@@ -11,6 +22,17 @@ describe('fromStream', () => {
       assert.equal(value, 1)
     }
   })
+  it('takes an old stream and returns an async iterable', async () => {
+    const stream = new PassThrough({ objectMode: true })
+    stream[Symbol.asyncIterator] = undefined as any
+    stream.write(1)
+    const itr = fromStream(stream)[Symbol.asyncIterator]()
+    assert.deepEqual(await itr.next(), { value: 1, done: false })
+    const next = itr.next()
+    stream.end()
+    assert.deepEqual(await next, { value: undefined, done: true })
+  })
+
   it('iterates empty streams', async () => {
     const stream = new PassThrough({ objectMode: true })
     stream.end()

lib/from-stream.ts

@@ -20,7 +20,7 @@ async function* _fromStream(stream: ReadableStreamish) {
       continue
     }
     if ((stream as any)._readableState.ended) {
-      return
+      break
     }
     await onceReadable(stream)
   }

lib/get-iterator-test.ts

@@ -46,4 +46,9 @@ describe('getIterator', () => {
       value: 1,
     })
   })
+  it("throws if it's not an iterator", () => {
+    assert.throw(() => {
+      getIterator(4 as any)
+    })
+  })
 })

lib/parallel-flat-map-test.ts

@@ -41,6 +41,15 @@ describe('parallelFlatMap', () => {
     }
     assert.deepEqual(values, ['1', '2', '3'])
   })
+  it('lets you curry a function and then fill the contract', async () => {
+    const values: any[] = []
+    const doubleTime = parallelFlatMap(2)
+    const stringParallelFlatMap = doubleTime(asyncStringArr, [1, 2, 3])
+    for await (const val of stringParallelFlatMap) {
+      values.push(val)
+    }
+    assert.deepEqual(values, ['1', '2', '3'])
+  })
   it('runs concurrent mapping operations', async () => {
     let mapCount = 0
     const counter = () => mapCount++

lib/reduce-test.ts

@@ -31,4 +31,9 @@ describe('reduce', () => {
     const addZero = addAble(0)
     assert.equal(await addZero([1, 2, 3]), 6)
   })
+  it('supports an undefined start after currying', async () => {
+    const reduceAdd = reduce((num1?: number, num2?: number) => (num1 || 0) + (num2 || 0))
+    const addZero = reduceAdd(undefined, [1, 2, 3])
+    assert.equal(await addZero, 6)
+  })
 })

lib/reduce.ts

@@ -19,7 +19,7 @@ export function reduce<T, B>(func: (acc: B, value: T) => B, start: B, iterable:
 export function reduce<T, B>(func: (acc: B, value: T) => B, start?: B, iterable?: AnyIterable<T>) {
   if (start === undefined) {
     return (curriedStart: B, curriedIterable?: AnyIterable<T>) =>
-      curriedIterable ? reduce(func, curriedStart, curriedIterable) : reduce(func, curriedStart)
+      curriedIterable ? _reduce(func, curriedStart, curriedIterable) : reduce(func, curriedStart)
   }
   if (iterable === undefined) {
     return (curriedIterable: AnyIterable<T>) => reduce(func, start, curriedIterable)

lib/take.ts

@@ -7,7 +7,7 @@ async function* _take<T>(count: number, iterable: AsyncIterable<T>) {
     yield await val
     taken++
     if (taken >= count) {
-      return
+      break
     }
   }
 }
@@ -18,7 +18,7 @@ function* _syncTake<T>(count: number, iterable: Iterable<T>) {
     yield val
     taken++
     if (taken >= count) {
-      return
+      break
     }
   }
 }

lib/write-to-stream-test.ts

@@ -12,6 +12,15 @@ describe('writeToStream', () => {
     assert.equal(stream.read(), 2)
     assert.equal(stream.read(), 3)
   })
+  it('curries', async () => {
+    const values = [1, 2, 3]
+    const stream = new PassThrough({ highWaterMark: 4, objectMode: true })
+    const streamWrite = writeToStream(stream)
+    await streamWrite(values)
+    assert.equal(stream.read(), 1)
+    assert.equal(stream.read(), 2)
+    assert.equal(stream.read(), 3)
+  })
   it('respects backpressure', async () => {
     let lastYield = 0
     function* values() {

lib/write-to-stream.ts

@@ -7,12 +7,6 @@ export interface WritableStreamish {
   removeListener: any
 }
 
-function once(event: string, stream: WritableStreamish): Promise<any> {
-  return new Promise(resolve => {
-    stream.once(event, resolve)
-  })
-}
-
 async function _writeToStream(stream: WritableStreamish, iterable: AnyIterable<any>): Promise<void> {
   let lastError = null
   let errCb: NullOrFunction = null