grpcToRx.ts

import { ClientReadableStream } from "grpc-web";
import { Observable } from "rxjs";

export function grpcToRx<T>(f: () => ClientReadableStream<T>): Observable<T> {
  return new Observable<T>(subscriber => {
    const stream = f();
    stream.on("error", e => subscriber.error(e));
    stream.on("end", () => subscriber.complete());
    stream.on("data", x => subscriber.next(x));
    return () => stream.cancel();
  });
}