syntax = "proto3";
package GrpcService;

service FileTransfer{
 rpc FileDownload (FileRequest) returns (stream FileReply);
 rpc FileUpload (stream FileReply) returns(stream FileReturn);
}

//请求下载文件时，所需下载文件的文件名称集合
message FileRequest{
 repeated string FileNames=1;//文件名集合
 //repeated重复字段 类似链表；optional可有可无的字段；required必要设置字段
 string Mark = 2;//携带的包
}

//下载和上传文件时的应答数据
message FileReply{
 string FileName=1;//文件名
 int32 Block = 2;//标记---第几个数据
 bytes Content = 3;//数据
 string Mark = 4;//携带的包
 }

//数据上传时的返回值
message FileReturn{
 string FileName=1;//文件名
 string Mark = 2;//携带的包
}

/// <summary>
/// 文件传输类
/// </summary>
class FileImpl:GrpcService.FileTransfer.FileTransferBase
{
 /// <summary>
 /// 文件下载
 /// </summary>
 /// <param name="request">下载请求</param>
 /// <param name="responseStream">文件写入流</param>
 /// <param name="context">站点上下文</param>
 /// <returns></returns>
 public override async Task FileDownload(FileRequest request, global::Grpc.Core.IServerStreamWriter<FileReply> responseStream, global::Grpc.Core.ServerCallContext context)
 {
 List<string> lstSuccFiles = new List<string>();//传输成功的文件
 DateTime startTime = DateTime.Now;//传输文件的起始时间
 int chunkSize = 1024 * 1024;//每次读取的数据
 var buffer = new byte[chunkSize];//数据缓冲区
 FileStream fs = null;//文件流
 try
 {
  //reply.Block数字的含义是服务器和客户端约定的
  for (int i = 0; i < request.FileNames.Count; i++)
  {
  string fileName = request.FileNames[i];//文件名
  string filePath = Path.GetFullPath($".//Files\\{fileName}");//文件路径
  FileReply reply = new FileReply
  {
   FileName = fileName,
   Mark = request.Mark
  };//应答数据
  Console.WriteLine($"{request.Mark}，下载文件：{filePath}");//写入日志，下载文件
  if (File.Exists(filePath))
  {
   fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, chunkSize, useAsync: true);

   //fs.Length 可以告诉客户端所传文件大小
   int readTimes = 0;//读取次数
   while (true)
   {
   int readSise = fs.Read(buffer, 0, buffer.Length);//读取数据
   if (readSise > 0)//读取到了数据，有数据需要发送
   {
    reply.Block = ++readTimes;
    reply.Content = Google.Protobuf.ByteString.CopyFrom(buffer, 0, readSise);
    await responseStream.WriteAsync(reply);
   }
   else//没有数据了，就告诉对方，读取完了
   {
    reply.Block = 0;
    reply.Content = Google.Protobuf.ByteString.Empty;
    await responseStream.WriteAsync(reply);
    lstSuccFiles.Add(fileName);
    Console.WriteLine($"{request.Mark}，完成发送文件：{filePath}");//日志，记录发送成功
    break;//跳出去
   }
   }
   fs?.Close();
  }
  else
  {
   Console.WriteLine($"文件【{filePath}】不存在。");//写入日志，文件不存在
   reply.Block = -1;//-1的标记为文件不存在
   await responseStream.WriteAsync(reply);//告诉客户端，文件状态
  }
  }
  //告诉客户端，文件传输完成
  await responseStream.WriteAsync(new FileReply
  {
  FileName = string.Empty,
  Block = -2,//告诉客户端，文件已经传输完成
  Content = Google.Protobuf.ByteString.Empty,
  Mark = request.Mark
  });
 }
 catch(Exception ex)
 {
  Console.WriteLine($"{request.Mark}，发生异常({ex.GetType()})：{ex.Message}");
 }
 finally
 {
  fs?.Dispose();
 }
 Console.WriteLine($"{request.Mark}，文件传输完成。共计【{lstSuccFiles.Count / request.FileNames.Count}】，耗时：{DateTime.Now - startTime}");
 }


 /// <summary>
 /// 上传文件
 /// </summary>
 /// <param name="requestStream">请求流</param>
 /// <param name="responseStream">响应流</param>
 /// <param name="context">站点上下文</param>
 /// <returns></returns>
 public override async Task FileUpload(global::Grpc.Core.IAsyncStreamReader<FileReply> requestStream, global::Grpc.Core.IServerStreamWriter<FileReturn> responseStream, global::Grpc.Core.ServerCallContext context)
 {
 List<string> lstFilesName = new List<string>();//文件名
 List<FileReply> lstContents = new List<FileReply>();//数据集合

 FileStream fs = null;
 DateTime startTime = DateTime.Now;//开始时间
 string mark = string.Empty;
 string savePath = string.Empty;
 try
 {
  //reply.Block数字的含义是服务器和客户端约定的
  while (await requestStream.MoveNext())//读取数据
  {
  var reply = requestStream.Current;
  mark = reply.Mark;
  if (reply.Block == -2)//传输完成
  {
   Console.WriteLine($"{mark}，完成上传文件。共计【{lstFilesName.Count}】个，耗时：{DateTime.Now-startTime}");
   break;
  }
  else if (reply.Block == -1)//取消了传输
  {
   Console.WriteLine($"文件【{reply.FileName}】取消传输！");//写入日志
   lstContents.Clear();
   fs?.Close();//释放文件流
   if (!string.IsNullOrEmpty(savePath) && File.Exists(savePath))//如果传输不成功，删除该文件
   {
   File.Delete(savePath);
   }
   savePath = string.Empty;
   break;
  }
  else if(reply.Block==0)//文件传输完成
  {
   if (lstContents.Any())//如果还有数据，就写入文件
   {
   lstContents.OrderBy(c => c.Block).ToList().ForEach(c => c.Content.WriteTo(fs));
   lstContents.Clear();
   }
   lstFilesName.Add(savePath);//传输成功的文件
   fs?.Close();//释放文件流
   savePath = string.Empty;

   //告知客户端，已经完成传输
   await responseStream.WriteAsync(new FileReturn
   {
   FileName= reply.FileName,
   Mark=mark
   });
  }
  else
  {
   if(string.IsNullOrEmpty(savePath))//有新文件来了
   {
   savePath = Path.GetFullPath($".//Files\\{reply.FileName}");//文件路径
   fs = new FileStream(savePath, FileMode.Create, FileAccess.ReadWrite);
   Console.WriteLine($"{mark}，上传文件：{savePath}，{DateTime.UtcNow.ToString("HH:mm:ss:ffff")}");
   }
   lstContents.Add(reply);//加入链表
   if (lstContents.Count() >= 20)//每个包1M，20M为一个集合，一起写入数据。
   {
   lstContents.OrderBy(c => c.Block).ToList().ForEach(c => c.Content.WriteTo(fs));
   lstContents.Clear();
   }
  }
  }
 }
 catch(Exception ex)
 {
  Console.WriteLine($"{mark}，发生异常({ex.GetType()})：{ex.Message}");
 }
 finally
 {
  fs?.Dispose();
 }
 }
}

class Program
{
 static void Main(string[] args)
 {
 //提供服务
 Server server = new Server()
 {
  Services = {GrpcService.FileTransfer.BindService(new FileImpl())},
  Ports = {new ServerPort("127.0.0.1",50000,ServerCredentials.Insecure)}
 };
 //服务开始
 server.Start();

 while(Console.ReadLine().Trim().ToLower()!="exit")
 {

 }
 //结束服务
 server.ShutdownAsync();
 }
}

/// <summary>
/// 传输结果
/// </summary>
/// <typeparam name="T"></typeparam>
class TransferResult<T>
{
 /// <summary>
 /// 传输是否成功
 /// </summary>
 public bool IsSuccessful { get; set; }
 /// <summary>
 /// 消息
 /// </summary>
 public string Message { get; set; }

 /// <summary>
 /// 标记类型
 /// </summary>
 public T Tag { get; set; } = default;
}

class FileTransfer
{

 /// <summary>
 /// 获取通信客户端
 /// </summary>
 /// <returns>通信频道、客户端</returns>
 static (Channel, GrpcService.FileTransfer.FileTransferClient) GetClient()
 {
 //侦听IP和端口要和服务器一致
 Channel channel = new Channel("127.0.0.1", 50000, ChannelCredentials.Insecure);
 var client = new GrpcService.FileTransfer.FileTransferClient(channel);
 return (channel, client);
 }

 /// <summary>
 /// 下载文件
 /// </summary>
 /// <param name="fileNames">需要下载的文件集合</param>
 /// <param name="mark">标记</param>
 /// <param name="saveDirectoryPath">保存路径</param>
 /// <param name="cancellationToken">异步取消命令</param>
 /// <returns>下载任务（是否成功、原因、失败文件名）</returns>
 public static async Task<TransferResult<List<string>>> FileDownload(List<string> fileNames, string mark, string saveDirectoryPath, System.Threading.CancellationToken cancellationToken = new System.Threading.CancellationToken())
 {
 var result = new TransferResult<List<string>>() { Message = $"文件保存路径不正确：{saveDirectoryPath}" };
 if (!System.IO.Directory.Exists(saveDirectoryPath))
 {
  return await Task.Run(() => result);//文件路径不存在
 }
 if (fileNames.Count == 0)
 {
  result.Message = "未包含任何文件";
  return await Task.Run(() => result);//文件路径不存在
 }
 result.Message = "未能连接到服务器";
 FileRequest request = new FileRequest() { Mark = mark };//请求数据
 request.FileNames.AddRange(fileNames);//将需要下载的文件名赋值
 var lstSuccFiles = new List<string>();//传输成功的文件
 string savePath = string.Empty;//保存路径
 System.IO.FileStream fs = null;
 Channel channel = null;//申明通信频道
 GrpcService.FileTransfer.FileTransferClient client = null;
 DateTime startTime = DateTime.Now;
 try
 {
  (channel, client) = GetClient();
  using (var call = client.FileDownload(request))
  {
  List<FileReply> lstContents = new List<FileReply>();//存放接收的数据
  var reaponseStream = call.ResponseStream;
  //reaponseStream.Current.Block数字的含义是服务器和客户端约定的
  while (await reaponseStream.MoveNext(cancellationToken))//开始接收数据
  {
   if (cancellationToken.IsCancellationRequested)
   {
   break;
   }
   if (reaponseStream.Current.Block == -2)//说明文件已经传输完成了
   {
   result.Message = $"完成下载任务【{lstSuccFiles.Count}/{fileNames.Count}】，耗时：{DateTime.Now - startTime}";
   result.IsSuccessful = true;
   break;
   }
   else if (reaponseStream.Current.Block == -1)//当前文件传输错误
   {
   Console.WriteLine($"文件【{reaponseStream.Current.FileName}】传输失败！");//写入日志
   lstContents.Clear();
   fs?.Close();//释放文件流
   if (!string.IsNullOrEmpty(savePath) && File.Exists(savePath))//如果传输不成功，删除该文件
   {
    File.Delete(savePath);
   }
   savePath = string.Empty;
   }
   else if (reaponseStream.Current.Block == 0)//当前文件传输完成
   {
   if (lstContents.Any())//如果还有数据，就写入文件
   {
    lstContents.OrderBy(c => c.Block).ToList().ForEach(c => c.Content.WriteTo(fs));
    lstContents.Clear();
   }
   lstSuccFiles.Add(reaponseStream.Current.FileName);//传输成功的文件
   fs?.Close();//释放文件流
   savePath = string.Empty;
   }
   else//有文件数据过来
   {
   if (string.IsNullOrEmpty(savePath))//如果字节流为空，则说明时新的文件数据来了
   {
    savePath = Path.Combine(saveDirectoryPath, reaponseStream.Current.FileName);
    fs = new FileStream(savePath, FileMode.Create, FileAccess.ReadWrite);
   }
   lstContents.Add(reaponseStream.Current);//加入链表
   if (lstContents.Count() >= 20)//每个包1M，20M为一个集合，一起写入数据。
   {
    lstContents.OrderBy(c => c.Block).ToList().ForEach(c => c.Content.WriteTo(fs));
    lstContents.Clear();
   }
   }
  }
  }
  fs?.Close();//释放文件流
  if (!result.IsSuccessful &&!string.IsNullOrEmpty(savePath)&& File.Exists(savePath))//如果传输不成功，那么久删除该文件
  {
  File.Delete(savePath);
  }
 }
 catch (Exception ex)
 {
  if (cancellationToken.IsCancellationRequested)
  {
  fs?.Close();//释放文件流
  result.IsSuccessful = false;
  result.Message = $"用户取消下载。已完成下载【{lstSuccFiles.Count}/{fileNames.Count}】，耗时：{DateTime.Now - startTime}";
  }
  else
  {
  result.Message = $"文件传输发生异常：{ex.Message}";
  }
 }
 finally
 {
  fs?.Dispose();
 }
 result.Tag = fileNames.Except(lstSuccFiles).ToList();//获取失败文件集合
 //关闭通信、并返回结果
 return await channel?.ShutdownAsync().ContinueWith(t => result);
 }


 /// <summary>
 /// 文件上传
 /// </summary>
 /// <param name="filesPath">文件路径</param>
 /// <param name="mark">标记</param>
 /// <param name="cancellationToken">异步取消命令</param>
 /// <returns>下载任务（是否成功、原因、成功的文件名）</returns>
 public static async Task<TransferResult<List<string>>> FileUpload(List<string> filesPath, string mark, System.Threading.CancellationToken cancellationToken=new System.Threading.CancellationToken())
 {
 var result = new TransferResult<List<string>> { Message = "没有文件需要下载" };
 if (filesPath.Count == 0)
 {
  return await Task.Run(() => result);//没有文件需要下载
 }
 result.Message = "未能连接到服务器。";
 var lstSuccFiles = new List<string>();//传输成功的文件
 int chunkSize = 1024 * 1024;
 byte[] buffer = new byte[chunkSize];//每次发送的大小
 FileStream fs = null;//文件流
 Channel channel = null;//申明通信频道
 GrpcService.FileTransfer.FileTransferClient client = null;
 DateTime startTime = DateTime.Now;
 try
 {
  (channel, client) = GetClient();
  using(var stream=client.FileUpload())//连接上传文件的客户端
  {
  //reply.Block数字的含义是服务器和客户端约定的
  foreach (var filePath in filesPath)//遍历集合
  {
   if(cancellationToken.IsCancellationRequested)
   break;//取消了传输
   FileReply reply = new FileReply()
   {
   FileName=Path.GetFileName(filePath),
   Mark=mark
   };
   if(!File.Exists(filePath))//文件不存在，继续下一轮的发送
   {
   Console.WriteLine($"文件不存在：{filePath}");//写入日志
   continue;
   }
   fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, chunkSize, useAsync: true);
   int readTimes = 0;
   while(true)
   {
   if (cancellationToken.IsCancellationRequested)
   {
    reply.Block = -1;//取消了传输
    reply.Content = Google.Protobuf.ByteString.Empty;
    await stream.RequestStream.WriteAsync(reply);//发送取消传输的命令
    break;//取消了传输
   }
   int readSize = fs.Read(buffer, 0, buffer.Length);//读取数据
   if(readSize>0)
   {
    reply.Block = ++readTimes;//更新标记，发送数据
    reply.Content = Google.Protobuf.ByteString.CopyFrom(buffer, 0, readSize);
    await stream.RequestStream.WriteAsync(reply);
   }
   else
   {
    Console.WriteLine($"完成文件【{filePath}】的上传。");
    reply.Block = 0;//传送本次文件发送结束的标记
    reply.Content = Google.Protobuf.ByteString.Empty;
    await stream.RequestStream.WriteAsync(reply);//发送结束标记
    //等待服务器回传
    await stream.ResponseStream.MoveNext(cancellationToken);
    if(stream.ResponseStream.Current!=null&&stream.ResponseStream.Current.Mark==mark)
    {
    lstSuccFiles.Add(filePath);//记录成功的文件
    }
    break;//发送下一个文件
   }
   }
   fs?.Close();
  }
  if (!cancellationToken.IsCancellationRequested)
  {
   result.IsSuccessful = true;
   result.Message = $"完成文件上传。共计【{lstSuccFiles.Count}/{filesPath.Count}】，耗时:{DateTime.Now - startTime}";

   await stream.RequestStream.WriteAsync(new FileReply
   {
   Block = -2,//传输结束
   Mark = mark
   }) ;//发送结束标记
  }
  }
 }
 catch(Exception ex)
 {
  if (cancellationToken.IsCancellationRequested)
  {
  fs?.Close();//释放文件流
  result.IsSuccessful = false;
  result.Message = $"用户取消了上传文件。已完成【{lstSuccFiles.Count}/{filesPath.Count}】，耗时:{DateTime.Now - startTime}";
  }
  else
  {
  result.Message = $"文件上传发生异常({ex.GetType()})：{ex.Message}";
  }
 }
 finally
 {
  fs?.Dispose();
 }
 Console.WriteLine(result.Message);
 result.Tag = lstSuccFiles;
 //关闭通信、并返回结果
 return await channel?.ShutdownAsync().ContinueWith(t => result);
 }
}

private string GetFilePath()
{
 // Create OpenFileDialog 
 Microsoft.Win32.OpenFileDialog dlg = new Microsoft.Win32.OpenFileDialog();

 // Set filter for file extension and default file extension 
 dlg.Title = "选择文件";
 dlg.Filter = "所有文件(*.*)|*.*";
 dlg.FileName = "选择文件夹.";
 dlg.FilterIndex = 1;
 dlg.ValidateNames = false;
 dlg.CheckFileExists = false;
 dlg.CheckPathExists = true;
 dlg.Multiselect = false;//允许同时选择多个文件 

 // Display OpenFileDialog by calling ShowDialog method 
 Nullable<bool> result = dlg.ShowDialog();

 // Get the selected file name and display in a TextBox 
 if (result == true)
 {
 // Open document 
 return dlg.FileName;
 }

 return string.Empty;
}
// 打开文件
private void btnOpenUpload_Click(object sender, RoutedEventArgs e)
{
 lblUploadPath.Content = GetFilePath();
}
CancellationTokenSource uploadTokenSource;
//上传
private async void btnUpload_Click(object sender, RoutedEventArgs e)
{
 lblMessage.Content = string.Empty;

 uploadTokenSource = new CancellationTokenSource();
 List<string> fileNames = new List<string>();
 fileNames.Add(lblUploadPath.Content.ToString());
 var result = await ServerNet.FileTransfer.FileUpload(fileNames, "123", uploadTokenSource.Token);

 lblMessage.Content = result.Message;

 uploadTokenSource = null;
}
//取消上传
private void btnCancelUpload_Click(object sender, RoutedEventArgs e)
{
 uploadTokenSource?.Cancel();
}


//打开需要下载的文件
private void btnOpenDownload_Click(object sender, RoutedEventArgs e)
{
 txtDownloadPath.Text = GetFilePath();
}
//下载文件
private async void btnDownload_Click(object sender, RoutedEventArgs e)
{
 lblMessage.Content = string.Empty;

 downloadTokenSource = new CancellationTokenSource();
 List<string> fileNames = new List<string>();
 fileNames.Add(System.IO.Path.GetFileName(txtDownloadPath.Text));
 var result= await ServerNet.FileTransfer.FileDownload(fileNames, "123", Environment.CurrentDirectory, downloadTokenSource.Token);

 lblMessage.Content = result.Message;

 downloadTokenSource = null;
}
CancellationTokenSource downloadTokenSource;
//下载取消
private void btnCancelDownload_Click(object sender, RoutedEventArgs e)
{
 downloadTokenSource?.Cancel();
}

您可能感兴趣的文章:

• golang在GRPC中设置client的超时时间
• golang 微服务之gRPC与Protobuf的使用
• golang grpc 负载均衡的方法
• 详解golang consul-grpc 服务注册与发现
• go grpc安装使用教程
• go实现grpc四种数据流模式

/*****PbTest.proto******/

syntax = "proto3";

package tutorial;

message Person {
  string name = 1;
  int32 id = 2;
  string email = 3;

  enum PhoneType {
    MOBILE = 0;
    HOME = 1;
    WORK = 2;
  }

  message PhoneNumber {
    string number = 1;
    PhoneType type = 2;
  }

  repeated PhoneNumber phones = 4;
}

message AddressBook {
   Person people = 1;
}

protoc --cpp_out=. PbTest.proto //先编译proto文件生成.cc和.h文件

#include <iostream>
#include "PbTest.pb.h"

void BookAdd(tutorial::Person *person)
{
  person->set_email("fun@qq.com");
  person->set_name("fun_name");
  person->set_id(1111);
  tutorial::Person::PhoneNumber *phone_num=person->add_phones();
  phone_num->set_number("1999");
}

int main() {
  tutorial::Person test;
  test.set_id(2111);
  test.set_name("main_name");
  test.set_email("main@qq.com");

  tutorial::Person::PhoneNumber phone_num;
  phone_num.set_number("2119");
  phone_num.set_type(tutorial::Person::WORK);

//  tutorial::Person_PhoneNumber phone;    //等价上面的phone_num
//  phone.set_number("2119");
//  phone.set_type(tutorial::Person_PhoneType::Person_PhoneType_HOME);
//  std::cout<<phone.number()<<std::endl;
//  std::cout<<phone.type()<<std::endl;

  tutorial::AddressBook book;
  BookAdd(book.mutable_people());
//  book.mutable_people()->set_name("main2test");    //与BookAdd函数调用等价

  std::cout<<"main id :"<<test.id()<<std::endl;
  std::cout<<"main name :"<<test.name()<<std::endl;
  std::cout<<"main email :"<<test.email()<<std::endl;
  std::cout<<"main phone :"<<phone_num.number()<<std::endl;
  std::cout<<"main phone type :"<<phone_num.type()<<std::endl;

  const tutorial::Person &person=book.people();
  std::cout<<"AddBook id :"<<person.id()<<std::endl;
  std::cout<<"AddBook name :"<<person.name()<<std::endl;
  std::cout<<"AddBook email :"<<person.email()<<std::endl;
  const tutorial::Person::PhoneNumber &num_phone=person.phones(0);
  std::cout<<"AddBook phone :"<<num_phone.number()<<std::endl;
  std::cout<<"AddBook phone type:"<<num_phone.type()<<std::endl;
  return 0;
}

g++ main.cc PbTest.pb.cc  -lprotobuf -lpthread

main id :2111
main name :main_name
main email :main@qq.com
main phone :2119
main phone type :2
AddBook id :1111
AddBook name :fun_name
AddBook email :fun@qq.com
AddBook phone :1999
AddBook phone type:0

var ByteBuffer = require("bytebuffer");
var ProtoBuf = require("protobuf");

        var protoFile = "protobuf/guaji";
        cc.loader.loadRes(protoFile,function (err,bgTexture2D){
            cc.log("loadfinish");

            Builder = ProtoBuf.protoFromString(bgTexture2D);
        });

        var Message = Builder.build("game.service.data.MsgLoginData");
        var loginMessage = new Message();
        loginMessage.set("UserAccount","webTest");
        loginMessage.set("UserPassWord","123456");

        var Message = Builder.build("game.service.data.MsgLoginResult");
        var msg = Message.decode(messageBuff);

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java   

项目：cfapi    文件：GeneratedMessageLite.java   

项目：cfapi    文件：GeneratedMessageLite.java   

项目：cfapi    文件：GeneratedMessageLite.java   

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java   

项目：play-store-api    文件：GeneratedMessageLite.java   

项目：bazel    文件：GeneratedMessageLite.java