StreamInterceptors do not allow async

[kurnal-volt]

UnaryInterceptors support async because they are promise base and it allows me to add authentication handling to my unary APIs. However, StreamInterceptors do not allow async so I cannot add authentication handling to my streaming APIs since getting the JWT using cognito is async.

[omrikiei]

@kurnal-volt I managed to get this done, but it ain't pretty :(

class SomeStreamInterceptor implements StreamInterceptor<any, any> {
    intercept(request: Request<any, any>,
        invoker: (request: Request<any, any>) =>
                  ClientReadableStream<any>): ClientReadableStream<any> {

        class InterceptedStream implements ClientReadableStream<any> {
            stream: ClientReadableStream<any>;
            constructor(stream: ClientReadableStream<any>) {
                this.stream = stream;
            }

            on(eventType: string, callback: any) {
                yourAsyncFunction().then(() => {
                    if (eventType == "data") {
                        this.stream.on("data", callback);
                    } else if (eventType == "error") {
                        this.stream.on("error", callback);
                    } else if (eventType == "metadata") {
                        this.stream.on("metadata", callback);
                    } else if (eventType == "status") {
                        this.stream.on("status", callback);
                    } else if (eventType == "end") {
                        this.stream.on("end", callback);
                    }
                });
                return this;
            }

            removeListener(eventType: string, callback: any) {
                if (eventType == "data") {
                    this.stream.removeListener("data", callback);
                } else if (eventType == "error") {
                    this.stream.removeListener("error", callback);
                } else if (eventType == "metadata") {
                    this.stream.removeListener("metadata", callback);
                } else if (eventType == "status") {
                    this.stream.removeListener("status", callback);
                } else if (eventType == "end") {
                    this.stream.removeListener("end", callback);
                }
            }
            cancel() { this.stream.cancel(); }
        }

        return new InterceptedStream(invoker(request));
    }
}