Nestjs / microservice不会创建可观察的吗？

Nestjs / microservice不会创建可观察的吗？"/>

Nestjs / microservice不会创建可观察的吗？

我正在尝试将流与grpc一起使用。但是，当我使用下面发布的代码时，它说subscribe不是函数。似乎nestjs / microservice不能从流中创建可观察对象。文档中使用的方法似乎已过时。

我的代码-原型文件：

message Rating {
  string id = 1;
  uint32 value = 2;
  string comment = 3;
  string type = 4;
}

service RatingsRpcService {
  rpc Test (stream GetRatingRequest) returns (stream Rating);
}

message GetRatingRequest {
  string id = 1;
}

message Ratings {
  repeated Rating items = 1;
}

和控制器文件：

    @GrpcStreamMethod('RatingsRpcService')
    test(msg: Observable<any>, metadata: any): Observable<any> {
        const subject = new Subject();
        msg.subscribe({
            next: (item: any) => {
                subject.next({whatever: 'value'});
            },
            error: (err: any) => console.log(err),
            complete: () => subjectplete()
        });
        return subject.asObservable();
    }

和我得到的错误：TypeError：msg.subscribe不是一个函数我错过了什么吗？

[好吧，我想我找到了问题。有点奇怪，但是从Observable更改类型可以修复该错误。我的意思是：messages: any，而不是消息：Observable<any>或消息：Observable<RatingInterface>。请找到下面显示的代码：

@GrpcStreamMethod('RatingsRpcService')
async findByIdStream(messages: any, metadata: any): Promise<Observable<RatingInterface>> {
    const subject = new Subject<RatingInterface>();
    messages.subscribe({
        next: async (dto: GetRatingDto) => {
            const item = await this.ratingsService.findById(dto.id);
            subject.next(item);
        },
        error: (err: any) => {
            throw new RpcException('Could not process stream.')
        },
        complete: () => subjectplete()
    });

    return subject.asObservable();
}