英文:
How can I implement a gRPC ServerWritableStream in nest.js?
问题
1 中的 [nest.js documentation] 没有提到有关单向 ServerWritableStream 的情况。我想要接收一个正常的请求，并使用 call.write 来传递给客户端流式消息。这在普通的 TypeScript 中可以正常工作，如下所示，但在 nest.js 的 gRPC 控制器中无法正常工作。我还在使用 Envoy，在 nest.js 上的一元调用以及简单服务器上也可以正常工作。
function doOnAdd(call) {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);call.write({message: myTodo});}, 5000);call.on('end', () => {console.log('end');});setTimeout(() => {call.end();}, 30000);}
但这个 nest.js 代码不起作用（同一控制器中的一元 gRPC 调用可以正常工作）。
@GrpcMethod('TodoService', 'OnAdded')async onAdded(request: todo.OnAddedTodoRequest,metadata: Metadata,call: ServerWritableStream<todo.OnAddedTodoRequest, todo.ServerMessage>,) {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);const message = new todo.ServerMessage({ message: myTodo });call.write(message);}, 5000);call.on('end', () => {console.log('end');});setTimeout(() => {call.end();}, 30000);}
最后，这是简化的 protobuf：
syntax = "proto3";package todo;service TodoService {rpc OnAdded (OnAddedTodoRequest) returns (stream ServerMessage);}message OnAddedTodoRequest {}message ServerMessage {string message = 1;}
英文:
The nest.js documentation doesn't mention anything regarding the case of a unidirectional ServerWritableStream. I want to receive a normal request and use call.write to pass to the client streaming messages. This works fine in plain TypeScript using the below but it doesn't work from within a nest.js gRPC controller. I am also using Envoy which works fine with the unary calls on nest.js as well as with the simple server.
function doOnAdd(call) {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);call.write({message: myTodo});}, 5000);call.on('end', () => {console.log('end');});setTimeout(() => {call.end();}, 30000);}
But this nest.js code does not work (unary gRPC calls work fine in the same controller).
@GrpcMethod('TodoService', 'OnAdded')async onAdded(request: todo.OnAddedTodoRequest,metadata: Metadata,call: ServerWritableStream<todo.OnAddedTodoRequest, todo.ServerMessage>,) {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);const message = new todo.ServerMessage({ message: myTodo });call.write(message);}, 5000);call.on('end', () => {console.log('end');});setTimeout(() => {call.end();}, 30000);}
Finally here is the simplified protobuf:
syntax = "proto3";package todo;service TodoService {rpc OnAdded (OnAddedTodoRequest) returns (stream ServerMessage);}message OnAddedTodoRequest {}message ServerMessage {string message = 1;}
答案1
得分: 0
我终于意识到 Nest.js 不等待间隔完成，而立即返回。我的解决方案是将其包装在一个Promise中，如下所示：
@GrpcMethod('TodoService', 'OnAdded')async onAdded(request: todo.OnAddedTodoRequest,metadata: Metadata,call: ServerWritableStream<todo.OnAddedTodoRequest, todo.ServerMessage>,) {await new Promise((resolve) => {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);const message = new todo.ServerMessage({ message: myTodo });call.write(message);}, 5000);call.on('end', () => {console.log('end');resolve();});setTimeout(() => {call.end();}, 30000);});}
英文:
I finally realised that nest.js doesn't wait for the interval to complete and returns immediately. My solution was to wrap it with a promise like this:
@GrpcMethod('TodoService', 'OnAdded')async onAdded(request: todo.OnAddedTodoRequest,metadata: Metadata,call: ServerWritableStream<todo.OnAddedTodoRequest, todo.ServerMessage>,) {await new Promise((resolve) => {setInterval(() => {const myTodo = JSON.stringify({id: 'b779cb10-72c8-416f-9399-273eab8e3421',title: 'Fix the server streaming',completed: false,});console.log('Sending streaming data', myTodo);const message = new todo.ServerMessage({ message: myTodo });call.write(message);}, 5000);call.on('end', () => {console.log('end');resolve();});setTimeout(() => {call.end();}, 30000);});}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论