javascript - gRPC 客户端流式传输 rpc 管道错误。(在结束错误后写入)
问题描述
我正在研究节点运行时的 gRPC 服务器-客户端编程。
我在客户端流式传输 rpc 中遇到错误。请参阅以下 rpc 方法签名。
service RouteGuide{
rpc DataStreaming(stream File) returns (Stats) {}
}
message Stats{
string msg=1;
}
message File{
bytes chk=1;
}
我想将文件从客户端上传到服务器。所以我定义了客户端流 rpc。
问题是文件上传只会第一次成功。
当我尝试上传另一个文件时,出现错误。在结束错误之后写。
我认为我没有很好地处理流。任何人都可以帮助为什么会发生这种情况?谢谢!
// server.js
"use strict";
const grpc = require("grpc");
const protoLoader = require("@grpc/proto-loader");
const path = require("path");
const fs = require("fs");
const stream = require("stream");
const PROTO_PATH = path.join(__dirname, "proto", "route.proto"); // path.resolve("proto", "route.proto")
const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
keepCase: false,
longs: String,
enums: String,
defaults: true,
oneofs: true
});
const routeguide = grpc.loadPackageDefinition(packageDefinition).routeguide;
const myTransformStream = new stream.Transform({
objectMode: true,
transform(data, enc, cb) {
cb(null, data.chk.toString());
}
});
function dataStreaming(strm, cb) {
console.log("server : streaming function");
stream.pipeline(
strm,
myTransformStream,
fs.createWriteStream("output.txt"),
err => {
if (err) {
console.log(`server side error : ${err}`);
cb(err);
} else {
console.log("server side no error");
cb(null, "server side finish");
}
}
);
}
function getServer() {
const server = new grpc.Server();
server.addService(routeguide.RouteGuide.service, {
DataStreaming: dataStreaming
});
return server;
}
if (require.main === module) {
const routeServer = getServer();
routeServer.bind("localhost:3333", grpc.ServerCredentials.createInsecure());
routeServer.start();
}
==============
// client.js
"use strict";
const grpc = require("grpc");
const protoLoader = require("@grpc/proto-loader");
const path = require("path");
const fs = require("fs");
const stream = require("stream");
const PROTO_PATH = path.join(__dirname, "proto", "route.proto"); // path.resolve("proto", "route.proto")
const packageDefinition = protoLoader.loadSync(PROTO_PATH, {
keepCase: false,
longs: String,
enums: String,
defaults: true,
oneofs: true
});
const routeguide = grpc.loadPackageDefinition(packageDefinition).routeguide;
const client = new routeguide.RouteGuide(
"localhost:3333",
grpc.credentials.createInsecure()
);
const MyTransform = new stream.Transform({
objectMode: true,
transform(chk, enc, cb) {
cb(null, { chk: chk });
}
});
function runDataStreaming() {
console.log("inside run-data-streaming()");
const strm = client.dataStreaming((err, ret) => {
if (err) {
console.log("client : file transfer failed.");
console.log(err);
} else {
console.log("client : file transfer succeeded.");
}
});
stream.pipeline(fs.createReadStream("test.txt"), MyTransform, strm, err => {
if (err) {
console.log(err.message);
} else {
console.log("pipeline succeeded");
}
});
}
if (require.main === module) {
runDataStreaming();
}
解决方案
我解决了我的问题。有一些问题。我会解释的。
myTransformStream
server.js 中的全局变量是问题所在。我将代码移动到dataStreaming
函数中。我还更改了关闭、销毁选项。详细地说,我打开了
autoclose
,emitclose
选项在fs.createReadStream()
,fs.createWriteStream()
和...打开emitclose
,autodestroy
选项在转换流(myTransformStream
和myTransform
)
推荐阅读
- c++ - 通过在编译时将指针与索引折叠来分配 C 数组
- swiftui - 如何访问另一个视图中的变量?
- php - mysql中是否有任何选项可以设置一列不应大于另一列
- node.js - 复制图像,读取其元数据,并使用流和锐化回写
- javascript - 如何使用 JS 将十六进制代码添加到 HTML 元素(内联样式)
- php - 将整数转换为字符串数组
- r - R:使用 dplyr 管道返回额外的 NULL
- python - WebHook 接收带有 +(加号)的字符串数据,其中空格应该是
- sql - “数据库 azure 数据中已经有一个名为 xy 的对象”
- xml - 允许元素来自一个 complexType AND/OR 另一个