Nest: gRPC client streaming call not working

[asmeikal]

Bug Report

Current behavior

Implementing a client streaming rpc call does not work with neither @GrpcStreamMethod or @GrpcStreamCall.

Using the first decorator results in a TypeError: call.write is not a function here.

Using the second decorator does not give any error, but the stream is never closed.

Input Code

Trying to implement a controller for this gRPC service is not currently possible.

syntax = "proto3";

package clientstream;

message ClientMessage {
  string message = 1;
}

message ResponseMessage {
  string response = 1;
}

service ClientStreamingService {
  rpc Rpc(stream ClientMessage) returns (ResponseMessage);
}

Expected behavior

It should be possible to implement a controller with a client streaming endpoint with one of the decorators provide by this framework.

Possible Solution

When using a client streaming rpc with @GrpcStreamCall, this method is called with a ServerReadableStream and a callback.

Changing the behavior of this decorator would break the integration tests, which assume all streaming is duplex.

A decorator for client streaming calls could be a solution, passing both the ServerReadableStream and the callback to the methodHandler.

export class ServerGrpc extends Server implements CustomTransportStrategy {
  /* ... */
  public createClientStreamCallMethod(methodHandler: Function) {
    return async (call: GrpcCall, callback: Function) => {
      methodHandler(call, callback);
    };
  }
  /* ... */
}

Environment

Nest version: 6.11.1

For Tooling issues:
- Node version: 10.8.0
- Platform:  Linux

[kamilmysliwiec]

Please, provide a minimal repository which reproduces your issue.

[asmeikal]

I set it up here.

I implemented both kind of streaming trying to follow the documentation.

To use the gRPC stream objects with these kind of streaming, one would need to rely on these types.

Something like this:

import { Controller } from '@nestjs/common';
import { GrpcClientStreamCall } from '@nestjs/microservices';
import { ServerReadableStream } from 'grpc';
import { clientstream } from '../codegen/server/clientstreaming';
import ClientMessage = clientstream.ClientMessage;
import ResponseMessage = clientstream.ResponseMessage;
import RpcWithCallCallback = clientstream.ClientStreamingService.RpcWithCallCallback;

@Controller()
export class ClientStreamingController {
  @GrpcClientStreamCall('ClientStreamingService', 'RpcWithCall')
  withCall(stream: ServerReadableStream<ClientMessage>, callback: RpcWithCallCallback): void {
    stream.on('data', msg => {
      /* process data */
    });

    stream.on('error', error => {
      callback(error);
    });

    stream.on('end', () => {
      /* create a response */
      callback(null, response);
    });
  }
}

The @GrpcClientStreamCall decorator in this example should create a service method like in my original message:

export class ServerGrpc extends Server implements CustomTransportStrategy {
  /* ... */
  public createClientStreamCallMethod(methodHandler: Function) {
    return async (call: GrpcCall, callback: Function) => {
      methodHandler(call, callback);
    };
  }
  /* ... */
}

[kamilmysliwiec]

Fixed + added docs + updated sample :) ref https://docs.nestjs.com/microservices/grpc#streaming-sample

[lock[bot]]

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.