Include a type parameter for error type in Stream

[TylorS]

Currently the types allow only the event values via Stream<A>. What if we also included the error type Stream<A, B>?

Reference TylorS/typed-stream#1
cc @goodmind

[briancavalier]

I think we could consider this now that we've clarified the distinction between application errors and stream failures. The new type parameter would be a stream failure type, rather than an application error type. For application errors, I think we should continue to point folks toward using an Either, which is already possible today:

const s: Stream<Either<string, MyCustomApplicationError>> = ...

vs a future where it's possible to be explicit about the stream failure type:

const s: Stream<Either<string, MyCustomApplicationError>, WebSocketClosedUnexpectedlyError> = ...

Those names are probably longer than what you'd typically encounter :) Even so, the type signature starts to get a little unwieldy--hopefully folks could lean on type inference most of the time.

We could potentially default the new type parameter to Error in both Flow and TS to avoid a breaking change, if we decide to do this before 2.0.

Thoughts?

[Frikki]

I like the long names.

[christianbradley]

I think we could consider this now that we've clarified the distinction between application errors and stream failures. The new type parameter would be a stream failure type, rather than an application error type. For application errors, I think we should continue to point folks toward using an Either, which is already possible today:

const s: Stream<Either<string, MyCustomApplicationError>> = ...

This seems a bit confusing to me... as it does not indicate the terminal nature of a stream exception.

Here's a general idea of the route I'd go...

class StreamFailure {
  private constructor(readonly cause: Error) {}
  static of: (e: Error) { return new StreamFailure(e) }
  static is(a: unknown): a is StreamFailure { return a instanceof StreamFailure }
}

type SinkException<A> = StreamFailure | A

interface Sink<L,A> {
  next(a: A): void
  error(e: SinkException<L>): void
  complete(): void
}

interface Stream<L,A> {
  run(sink: Sink<L,A>): Disposable
}

const foldSinkException = <A,B>(
  f: (e: StreamFailure) => B,
  g: (e: A) => B
) => (e: SinkException<A>) => StreamFailure.is(e) ? f(e) : g(e)

// example usage
type AppException = 
  | { tag: 'tag1', data: string }
  | { tag: 'tag2', data: number }

const myStream: Stream<AppException, void> = getMyStream()

myStream.run({
  next: () => console.log("tick"),
  error: foldSinkException<AppException, void>(
    l => console.error(l.cause),
    a => {
      switch(e.tag) {
        case 'tag1': return console.log(e.data)
        case 'tag2': return console.log(e.data.toString())
      }
    }
  ),
  complete: () => console.log('done')
})