syntax = "proto3";
//引入集合包
import "google/protobuf/empty.proto";
//命名空間
option csharp_namespace = "GrpcDemo";
//包名
package MyTest;
//接口定義
service MyTestDemo {
  rpc MultipleParam(MultipleRequestPara) returns (MultipleRespone);
  rpc NoParam(google.protobuf.Empty) returns (SingeRespone);
  rpc CollectionParam(google.protobuf.Empty) returns (CollectionResponePara);
}

//多參數(shù)請求參數(shù)
message MultipleRequestPara {
  int32 Id = 1;
  string Name = 2;//參數(shù)個數(shù)
  bool IsExists =3;
}
message SingeRespone {
  bool Success =1;
  TestEntity a1 = 2;
  message TestEntity{
	int32 Id =1;
  }
}
//多參數(shù)返回
message MultipleRespone {
	bool Success =1;
}
//返回集合參數(shù)
message CollectionResponePara {
	repeated CollectionChildrenRespone1 param1 =1;
	repeated CollectionChildrenRespone2 param2 =2;
	repeated int32 param3 =3;
}
//集合屬性1
message CollectionChildrenRespone1 {
	int32 Id =1;
}
//集合屬性2
message CollectionChildrenRespone2 {
	string Name =1;	
}

/// <summary>
/// 繼承自MyTestDemo.MyTestDemoBase
/// </summary>
public class MyTestService : MyTestDemo.MyTestDemoBase
{
    public override async Task<MultipleRespone> MultipleParam(MultipleRequestPara request, ServerCallContext context)
    {
        return await Task.FromResult(new MultipleRespone
        {
            Success = true,
        });
    }

    public override async Task<SingeRespone> NoParam(Empty request, ServerCallContext context)
    {
       TestEntity t = new TestEntity();
       t.Id = 1;
       return await Task.FromResult(new SingeRespone { Success = true, entity = t  }); ;
    }

    public override async Task<CollectionResponePara> CollectionParam(Empty request, ServerCallContext context)
    {
        CollectionResponePara collectionResponePara = new CollectionResponePara();
        CollectionChildrenRespone1 a = new CollectionChildrenRespone1 { Id = 1 };
        CollectionChildrenRespone2 b = new CollectionChildrenRespone2 { Name = "jeck" };
        collectionResponePara.Param1.Add(a);
        collectionResponePara.Param2.Add(b);
        return  await  Task.FromResult(collectionResponePara);
    }
}

using (var channel = GrpcChannel.ForAddress("https://localhost:7245"))
 {
     var client =  new MyTestDemo.MyTestDemoClient(channel);
     //多參數(shù)調(diào)用
     var reply = client.MultipleParam(new MultipleRequestPara { Id = 123, Name = "sa", IsExists = true });  
     //無參調(diào)用
     var singeRespone = client.NoParam(new Google.Protobuf.WellKnownTypes.Empty());
     //調(diào)用集合
     var collectionResponePara = client.CollectionParam(new Google.Protobuf.WellKnownTypes.Empty());
 }

syntax = "proto3";
option csharp_namespace = "GrpcDemo";
package streamtest;

service StreamTest {
  //服務(wù)端流定義
  rpc ExcuteServerStream(StreamForClientRequest) returns (stream StreamForClientRespones);
  //客戶端流定義
  rpc ExcuteServerStream(StreamForClientRequest) returns (stream StreamForClientRespones);
  //雙向流
  rpc ExcuteMutualStream(stream StreamForClientRequest) returns ( stream StreamForClientRespones);
}

//調(diào)用流的請求對象
message StreamForClientRequest{
    int32 Id=1;
}

//調(diào)用端流的返回對象
message StreamForClientRespones{
	repeated int32 Number=1;//集合
}

//服務(wù)端流接口
public override async Task ExcuteServerStream(StreamForClientRequest req,IServerStreamWriter<StreamForClientRespones> resStream,ServerCallContext context)
{
    //list集合作為模擬數(shù)據(jù)源
    var list = new List<int> { 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13 };
    foreach (var item in list)
    {
        Console.WriteLine($"********{item}*******");
        var ele = new StreamForClientRespones();
        ele.Number.Add(item);
        //寫入流中
        await resStream.WriteAsync(ele);
        //模擬源源不斷的數(shù)據(jù)響應(yīng)
        await Task.Delay(1000);
    }
}

//客戶端流接口
public override async Task<StreamForClientRespones> ExcuteClientStream( IAsyncStreamReader<StreamForClientRequest> requestStream, ServerCallContext context)
{
    StreamForClientRespones intArrayModel = new StreamForClientRespones();
    //獲取請求流中的數(shù)據(jù)
    while (await requestStream.MoveNext())
    {
        intArrayModel.Number.Add(requestStream.Current.Id + 1);
        Console.WriteLine($"ExcuteClientStream Number {requestStream.Current.Id} 獲取到并處理.");
        Thread.Sleep(100);
    }
    return intArrayModel;
}

//雙向流
public override async Task ExcuteMutualStream(IAsyncStreamReader<StreamForClientRequest> reqStream,IServerStreamWriter<StreamForClientRespones> resStream,ServerCallContext context)
{
    int i = 0;
    //從流中獲取請求
    while (await reqStream.MoveNext())
    {
        i++;
        var ele = new StreamForClientRespones();
        ele.Number.Add(i);
        //寫入響應(yīng)流
        await resStream.WriteAsync(ele);
        await Task.Delay(500);
    }
}

//調(diào)用服務(wù)端流 
using (var channel = GrpcChannel.ForAddress("https://localhost:7245"))
 {
     var client = new StreamTest.StreamTestClient(channel);
     //調(diào)用服務(wù)端流
     var reply =  client.ExcuteServerStream(new StreamForClientRequest { Id =1});

     //利用線程取消
     //CancellationTokenSource cts = new CancellationTokenSource();
     //指定在2s后進行取消操作
     //cts.CancelAfter(TimeSpan.FromSeconds(2.5)); 
     //var reply = client.ExcuteServerStream(new StreamForClientRequest { Id = 1 }, cancellationToken: cts.Token);

     await foreach (var resp in reply.ResponseStream.ReadAllAsync())
     {
         Console.WriteLine(resp.Number[0]);
     }
 }

 //調(diào)用客戶端流
 using (var channel = GrpcChannel.ForAddress("https://localhost:7245"))
 {
     var client = new StreamTest.StreamTestClient(channel);
     //調(diào)用客戶端流接口
     var reply = client.ExcuteClientStream();

     //模擬源源不斷的數(shù)據(jù)發(fā)送
     for (int i = 0; i < 10; i++)
     {
         await reply.RequestStream.WriteAsync(new StreamForClientRequest() { Id = new Random().Next(0, 20) });
         await Task.Delay(100);
     }
     Console.WriteLine("*************發(fā)送完畢*******************");
     await reply.RequestStream.CompleteAsync();
     //接受結(jié)果
     foreach (var item in reply.ResponseAsync.Result.Number)
     {
         Console.WriteLine($"This is {item} Result");
     }
 }

//雙向流
using (var channel = GrpcChannel.ForAddress("https://localhost:7245"))
{
    var client = new StreamTest.StreamTestClient(channel);
    //調(diào)用雙向流接口
    var reply = client.ExcuteMutualStream();
    //獲取流放入線程
    var bathCatRespTask = Task.Run(async () =>
    {
        await foreach (var resp in reply.ResponseStream.ReadAllAsync())
        {
            Console.WriteLine(resp.Number[0]);
        }
    });

    //寫入流
    for (int i = 0; i < 10; i++)
    {
        await reply.RequestStream.WriteAsync(new StreamForClientRequest() { Id = new Random().Next(0, 20) });
        await Task.Delay(100);
    }
    //發(fā)送完畢
    await reply.RequestStream.CompleteAsync();
    //開始接收響應(yīng)
    await bathCatRespTask;
}

builder.Services.AddGrpcClient<MyTestDemo.MyTestDemoClient>(option => {
    option.Address = new Uri("https://localhost:7245");
});

 public class gRPCTestController : ControllerBase
 {
     private readonly MyTestDemoClient _client;
     public gRPCTestController(MyTestDemoClient client)
     {
         _client = client;
     }

     [HttpGet(Name = "Excute")]
     public async Task<string> Get()
     {
         var a = await _client.NoParamAsync(new Google.Protobuf.WellKnownTypes.Empty());
         var str = a.Success.ToString();
         return str;
     }
 }

public class UnaryServerHandlerInterceptor : Interceptor
{
    public override async Task<TResponse> UnaryServerHandler<TRequest, TResponse>(
       TRequest request,
       ServerCallContext context,
       UnaryServerMethod<TRequest, TResponse> continuation)
    {
        Console.WriteLine("執(zhí)行調(diào)用前");
        var result = await continuation(request, context);
        Console.WriteLine("執(zhí)行調(diào)用后");
        // 或向 客戶端附加 一些信息
        // 也可以 用try catch 做異常日志
        // 可以從 context中取出 調(diào)用方ip，做ip限制
        // 可以 監(jiān)控continuation 的 執(zhí)行時間
        return result;
    } 
}

builder.Services.AddGrpc(option => { 
    option.EnableDetailedErrors = true;
    //加入服務(wù)端攔截器選項
    option.Interceptors.Add<UnaryServerHandlerInterceptor>();
});