feat: Convert the library to typescript (#27)

MAJOR changed the push type signature to return boolean, a breaking change but makes it more correct

.npmignore

@@ -1,8 +1,9 @@
 package-lock.json
 .nvmrc
 test
-.babelrc
 .gitignore
 .travis.yml
 lib
 .vscode
+tsconfig.json
+tslint.json

lib/collect.js → lib/collect.ts

@@ -1,9 +1,9 @@
-import { write } from './write'
 import { pipe } from './pipe'
+import { write } from './write'
 
 export async function collect (stream) {
   const acc = []
-  await pipe(stream, write(data => acc.push(data)))
+  await pipe(stream, write(data => { acc.push(data) }))
   if (acc.length === 0) {
     return null
   }

lib/filter.ts

@@ -0,0 +1,25 @@
+import { ITransformStreamOptions, TransformStream } from './transform'
+
+export type filterFunc = (data: any, encoding?: string) => boolean | Promise<boolean>
+
+export class FilterStream extends TransformStream {
+  private filterFunction: filterFunc
+
+  constructor (opts: ITransformStreamOptions | filterFunc, filterFunction?: filterFunc) {
+    if (typeof opts === 'function') {
+      filterFunction = opts
+      opts = {}
+    }
+    super(opts)
+    this.filterFunction = filterFunction
+  }
+
+  public async _transform (data, encoding) {
+    const keep = await Promise.resolve(this.filterFunction(data, encoding))
+    if (keep) {
+      await this.push(data)
+    }
+  }
+}
+
+export const filter = (opts: ITransformStreamOptions | filterFunc, fn?: filterFunc) => new FilterStream(opts, fn)

lib/index.ts

@@ -0,0 +1,58 @@
+import { collect } from './collect'
+import { filter, FilterStream } from './filter'
+import { pipe } from './pipe'
+import { IReadableStreamOptions, read, readFunction, ReadStream } from './read'
+import { readAsync } from './readAsync'
+import { reduce, reduceFunction, ReduceStream } from './reduce'
+import { ITransformStreamOptions, map, transform, transformFunction, TransformStream } from './transform'
+import { wait } from './utils'
+import { IWritableStreamOptions, write, writeFunction, WriteStream } from './write'
+
+// types
+export { IBluestream } from './interfaces'
+export {
+  IReadableStreamOptions,
+  IWritableStreamOptions,
+  ITransformStreamOptions,
+  transformFunction,
+  readFunction,
+  reduceFunction,
+  writeFunction,
+}
+
+// methods
+export {
+  FilterStream,
+  ReadStream,
+  ReduceStream,
+  TransformStream,
+  WriteStream,
+  collect,
+  filter,
+  map,
+  pipe,
+  read,
+  readAsync,
+  reduce,
+  transform,
+  wait,
+  write,
+}
+
+export default {
+  FilterStream,
+  ReadStream,
+  ReduceStream,
+  TransformStream,
+  WriteStream,
+  collect,
+  filter,
+  map,
+  pipe,
+  read,
+  readAsync,
+  reduce,
+  transform,
+  wait,
+  write,
+}

lib/interfaces.ts

@@ -0,0 +1,5 @@
+import { Readable, Stream } from 'stream'
+
+export interface IBluestream extends Stream {
+  promise (): Promise<void>
+}

lib/pipe.js → lib/pipe.ts

@@ -1,7 +1,8 @@
+import { Readable, Writable } from 'stream'
 import { defer } from './utils'
 
-export async function pipe () {
-  const streams = Array.from(arguments)
+export async function pipe (readable: Readable, ...writableStreams: Writable[]) {
+  const streams = [readable, ...writableStreams]
   if (streams.length < 2) {
     throw new TypeError('Must pipe to two or more streams')
   }
@@ -13,7 +14,7 @@ export async function pipe () {
     const lastStream = index + 1 === streams.length
     if (!lastStream) {
       const nextStream = streams[index + 1]
-      stream.pipe(nextStream)
+      stream.pipe((nextStream as Writable))
     }
   })