GRPC Proto Server Bi-directional streaming controller incorrect behavior — doesn't provide stream handler back to method controller, while providing undefined reference

[anton-alation]

I'm submitting a...

[ ] Regression 
[X] Bug report
[ ] Feature request
[ ] Documentation issue or request
[ ] Support request => Please do not submit support request here, instead post your question on Stack Overflow.

Current behavior

Defined following proto

service Service {
    rpc ObserveStream(stream Message) returns (stream Message);
}

Defined following controller

@Controller()
export class ObserveController implements OnModuleInit {

    @GrpcMethod("Service")
    public observeStream(stream: any, meta?: any): any {
          console.log(arguments);
    }

}

Console log on arguments outputs:

[Arguments] {
  '0': undefined,
  '1': Metadata { _internal_repr: { 'user-agent': [Array] } } }

Expected behavior

First parameter should be stream handler with .on and .write methods available for execution

Possible problem

Stream service definitions doesn't have .request option on them, and should be passed down to clients as transport handlers with .on and .write methods available.

Possible fix: in node_modules/@nestjs/microservices/server/server-grpc.js
replace call.request to call on line 76

createStreamServiceMethod(methodHandler) {
        return async (call, callback) => {
            const handler = methodHandler(call, call.metadata);
            const result$ = this.transformToObservable(await handler);
            await result$
                .pipe(operators_1.takeUntil(rxjs_1.fromEvent(call, constants_1.CANCEL_EVENT)))
                .forEach(data => call.write(data));
            call.end();
        };
    }

Same fix for line: https://github.com/nestjs/nest/blob/5.4.0/packages/microservices/server/server-grpc.ts#L114

Minimal reproduction of the problem with instructions

Define bi-directional streaming GRPC interface via protobuffers and then create GRPC controller implementation as above

Environment

Nest version: 5.4.0

 
For Tooling issues:
- Node version: v10.9.0
- Platform:  Mac

Others:
OSX 10.13
Webstorm

[adubrov]

@anton-alation I created similar issue - #1264

[anton-alation]

@adubrov I made it to PR :) #1292

Kinda I need this feature like yesterday, or I need to replace a framework with the custom-made framework, which isn't a really productive way to build things :)

[adubrov]

@anton-alation,
Oh, cool. I was thinking about it too, but was kinda busy.
I need it yesterday as well )) but I decided to use nest for non-stream methods, and when I need to stream - there is a dirty hack - you can pick Grpc server directly from nest microservice and use it.

[adubrov]

@anton-alation, I've looked at your pr. Isn't it necessary to change grpc-client as well? To allow to send Observable and Subject?

[anton-alation]

@adubrov So we have a bit of cases here:

Working

Client Unary -> Server Unary
Client Unary -> Server Stream

Not Working

Client Stream -> Server Unary
Client Stream -> Server Stream

So in the big picture, GRPC support on Nest is working only 50% of it should be actually.

As we can assume (I am not big user of Nest, but bringing it to life for one of the projects) the GRPC-Client is a way to dispatch messages to Grpc-server within Alation, so particular use-cases can be:

HTTP -> GRPC-CLIENT -> GRPC-SERVER

In this case, we can assume that there can be no streaming because we'll treat HTTP as a single dispatch message system, otherwise there probably should be WS or GRPC implementation for cases which require messaging

WS-CLIENT -> WS-SERVER -> GRPC-CLIENT -> GRPC-SERVER -> GRPC-CLIENT -> WS-SERVER -> WS-CLIENT

In this case, we need to assume that WS-SERVER as well can dispatch to WS-CLIENT back, and this is what actually supported on Nest WS implementation: https://docs.nestjs.com/websockets/gateways

Assuming that between all of the parts we have networking abstraction there is none of the problems should happen to GRPC client because it's creating same stream object on it's own side awaiting stream to dispatch to it, providing Subscribable object back.
While on the server it's would be just common node stream object you can do .on and .write.

One big thing we lack is a supporting documentation and code commentary for Nest GRPC.

Second big thing is not enough understanding why it's required to be Observable (or Promise) passed back every time, and no actual stream passed to be dispatched for manual control (because we may have mechanics to dispatch that other than default Nest), so it seems very crucial to be able to get control over the stream with signaling to Nest that it shouldn't make any attempts to control the stream.

It's a matter of choice on how to build specific applications which require persistent connections to be strict part of it, and Observable pattern (and RxJs) can actually complicate things there neither than provide any help :)

Subjective of course ;)

[adubrov]

@anton-alation it not alway only this cases you described. Sometimes it just interaction between 2 microservices when you need to have streaming.
I'm agree that it better to have a way to expose raw stream, but for me it much more convenient to operate with Observable transaction then with events/callbacks from stream.

[anton-alation]

@adubrov PR doesn't prevent you to continue with Observable in the way you want to use it. Instead of returning null from handler just return an Observable/Promise and you'll good to go ;) Both worlds happy, nobody blocked!

My problem is that the stream handler in my case going into the very different dispatcher than a controller we catching the stream in, so I really need the stream handler for very intelligent mechanics ;)

Anyway check this line:
https://github.com/nestjs/nest/pull/1292/files#diff-ad4ad5b901e205c42627f37db1ffd90dR131

It's explicitly said that if you want observables you got those in full ;)

[kamilmysliwiec]

Published as 6.3.0

[lock]

This thread has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.