NestJS gRPC Unimplemented Streaming Server Method

Question

I'm trying to build microservice with NestJS and gRPC. It has two services, the first one is the gRPC service, and the second one is the REST service that call gRPC service.

At first, it works fine on unary call, the findRepoById rpc. But it doesn't work on server streaming call, findAllRepos rpc. It throws error like this when i tried to call findAllRepos rpc

UnhandledPromiseRejectionWarning: Error: 12 UNIMPLEMENTED: The server does not implement the method findAllRepos

I wrote the files as shown below

// main.proto

syntax = "proto3";

import "google/protobuf/timestamp.proto";

package main;

enum Visibility {
  PUBLIC = 0;
  PRIVATE = 1;
}

message Repository {
  int32 id = 1;
  string title = 2;
  Visibility visibility = 3;
  google.protobuf.Timestamp lastSeen = 4;
}

service MainService {
  rpc findRepoById (RepoById) returns (Repository) {}
  rpc findAllRepos (NoParam) returns (stream Repository) {}
}

message RepoById {
  int32 id = 1;
}

message NoParam {}

// server.controller.ts

@Controller('repo')
export class RepoController {
  constructor(
    @Inject('RepoService') private readonly repoService: RepoService
  ) {}

  @GrpcMethod('MainService', 'findRepoById')
  findRepoById(request: RepoById, metadata: Metadata): Repository {
    const targetId = request.id;
    return this.repoService.findRepoById(targetId);
  }

  @GrpcStreamMethod('MainService', 'findAllRepos')
  findAllRepos(request: NoParam, metadata: Metadata): Observable<Repository> {
    const subject = new Subject<Repository>();

    const repositories = this.repoService.findAllRepos();
    repositories.map((repo) => {
      subject.next(repo);
    });
    subject.complete();

    return subject.asObservable();
  }
}

// client.service.ts

export class RepoGrpcService implements OnModuleInit {
  private mainService: MainServiceClient;

  constructor(@Inject('main_package') private client: ClientGrpc) {}

  onModuleInit() {
    this.mainService = this.client.getService<MainServiceClient>('MainService');
  }

  findRepoById(id: number): Observable<Repository> {
    return this.mainService.findRepoById({ id });
  }

  @GrpcStreamCall('MainService')
  findAllRepos(): Observable<Repository[]> {
    const results: Repository[] = [];

    const repoStream = this.mainService.findAllRepos({});
    repoStream.forEach((value) => console.log(value));
    repoStream.subscribe({
      next: (repo) => {
        results.push(repo);
      }
    });

    const subject = new Subject<Repository[]>();
    subject.next(results);
    subject.complete();

    return subject.asObservable();
  }
}

I think I already followed all the code same as on NestJS gRPC documentation But somehow it still doesn't work. Is there something wrong I do?

Answer 1

I know it's been a long time since this question was asked, but maybe this answer will help others. Please Attention to documentation, some parts of the code are not mentioned because of previous chapters, So NestJS is not an Exception. It seems, that your problem comes from the controller-module relation Take a look at the documentation again and be sure that your controller is defined in the correct module and your module relations(import) are fine.

NestJs Doc-Module

Official Example

Answer 2

From my experience if it's server side streaming, the @GrpcMethod annotation works. Seems the document doesn't describe this bit clearly enough, but to me it sounds like only mutual stream services needs GrpcStreamMethod?