介绍

一个高性能、开源和通用的 RPC 框架，基于ProtoBuf(Protocol Buffers) 序列化协议开发，且支持众多开发语言。面向服务端和移动端，基于 HTTP/2 设计，带来诸如双向流、流控、头部压缩、单 TCP 连接上的多复用请求等特性。这些特性使得其在移动设备上表现更好，更省电和节省空间占用。

特性

• 🚀 rpc4cj可以跨语言使用。支持多种语言 支持C++、Java、Go、Python、Ruby、C#、Node.js、Android Java、Objective-C、PHP等编程语言。

• 🚀 基于 IDL ( 接口定义语言（Interface Define Language）)文件定义服务，通过 proto3 工具生成指定语言的数据结构、服务端接口以及客户端 Stub。

• 💪 通信协议基于标准的 HTTP/2 设计，支持双向流、消息头压缩、单 TCP 的多路复用、服务端推送等特性，这些特性使得 rpc4cj 在移动端设备上更加省电和节省网络流量。

• 🛠️ 序列化支持 PB（Protocol Buffer），PB 是一种语言无关的高性能序列化框架，基于 HTTP/2 + PB, 保障了 RPC 调用的高性能。

• 🌍 安装简单，扩展方便（用该框架每秒可达到百万个RPC）。

软件架构

架构图

• rpc4cj使客户端和服务端应用程序可以透明地进行通信,并简化了连接系统的构建。

源码目录

.
├── README.md
├── doc
│   ├── assets
│   ├── cjcov
│   ├── feature_api.md
│   └── design.md
├── src
│   └── attributes
│   └── buffer
│   └── exceptions
│   └── grpc
│   └── internal
│   └── net
│   └── resolver
│   │   └── dns
│   │   └── manual
│   │   └── passthrough
│   └── transport
│   └── util     
└── test
    ├── HLT
    ├── LLT
    └── UT

• doc存放库的设计文档、提案、库的使用文档、LLT 用例覆盖报告
• src是库源码目录
• test存放测试用例，包括HLT用例、LLT 用例和UT用例

接口说明

主要是核心类和成员函数说明，详情见 API

使用说明

编译

编译描述和具体 shell 命令

linux x86 编译：

cjpm build

Windows x86 编译：

前提：使用 openssl version 命令查看环境中是否已安装 openssl，版本需为 OpenSSL 3.x.x。若未安装，可参考 https://github.com/openssl/openssl 官网的 Build and Install 部分。

cjpm build

功能示例

需要安装 protobuf 软件及 protobuf4cj 插件： ubuntu 系统可使用以下命令安装 protobuf

apt install -y protobuf-compiler
protoc --version # 正确显示版本号即成功

编译安装 protobuf-gen-cj 插件，可在 https://gitcode.com/Cangjie-TPC/protobuf4cj.git 处编译获取

使用 protobuf4cj 用该 proto 文件生成 helloworld_pb.cj 文件，放入 test/LLT 中，可以使用以下命令，参考 README.md · Cangjie-TPC/protobuf4c - 码云 - 开源中国 (gitcode.com)

protoc --cj_out=.  --proto_path=. $(proto_files) #--cj_out= 指定输出地址，--proto_path=指定proto文件路径

一元调用功能示例

首先运行任意一版本语言的 helloworld server example，或本仓附带 go 版本 server 端 其他语言版本 server 端可于 Supported languages | gRPC 内获取

示例 proto 文件如下：

syntax = "proto3";

option go_package = "google.golang.org/grpc/examples/helloworld/helloworld";
option java_multiple_files = true;
option java_package = "io.grpc.examples.helloworld";
option java_outer_classname = "HelloWorldProto";

package helloworld;

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

示例代码如下：

使用测试框架运行，将 “xxxx" 处替换为本地代码仓库地址，并将源代码文件放入test/LLT中。

// DEPENDENCE: ./helloworld.pb.cj
// EXEC: cjc %import-path %L %l %f ./helloworld.pb.cj
// EXEC: ./main

public import std.net.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import std.collection.*
public import protobuf.*
public import std.time.*

var port: UInt16 = 50051
main (){
    var req = HelloRequest()
    var resp = HelloReply()
    req.name = "World"
    var con = dial("127.0.0.1:${port}")
    con.invoke("/helloworld.Greeter/SayHello", req, resp)
    println("result: ${resp}")
}

启动仓内提供测试用 go 版本服务器

./test/LLT/test_grpc_post/go_server

另开一终端执行命令

python3 ci_test/main.py test

执行结果如下：

result: {message: "Hello World"}

流式调用功能示例

cangjie 双向流式调用的示例如下:

示例代码如下：

public import std.socket.*
public import std.io.*
public import std.collection.*
public import stdx.net.tls.*
public import stdx.net.tls.common.*
public import crypto.x509.*
public import stdx.crypto.keys.*
public import stdx.crypto.common.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import protobuf.*
public import std.sync.*
public import std.time.*
public import std.fs.*


let strServer: String = "test"
var port: UInt16 = 0
let num: Int64 = 5
class ServerTest <: GreeterServer {
    // SayHello implements helloworld.GreeterServer
    public func SayHello(_: HelloRequest): (HelloReply, GrpcError) {
        let hy: HelloReply = HelloReply()
        hy.message = strServer
        return (hy, GrpcError())
    }
    public func mustEmbedUnimplementedGreeterServer() {}
}
main(): Int64 {    
    let pem = String.fromUtf8(File("./example-cert.pem", OpenOption.Open(true, false)).readToEnd())
    let key = String.fromUtf8(File("./example-key.pem", OpenOption.Open(true, false)).readToEnd())
    var cfg_temp = TlsServerConfig(X509Certificate.decodeFromPem(pem),GeneralPrivateKey.decodeFromPem(key))
    cfg_temp.supportedAlpnProtocols = ["h2"]
    let ss: TcpServerSocket = TcpServerSocket(bindAt: port)
    ss.bind()
    let cfgs = cfg_temp
    port = (ss.localAddress as IPSocketAddress).getOrThrow().port
    
    spawn{ =>
        let serve = TlsSocket.server(ss.accept(),serverConfig: cfgs)
        let lis: GRPCServerSocket = GRPCServerSocket(GrpcTlsS(serve))
        let server: Server = Server()
        registerGreeterServer(server, ServerTest())
        server.serve(lis)
    }

    sleep(Duration.second)

    var req = HelloRequest()
    var resp = HelloReply()
    req.name = "World"
    var cfg = TlsClientConfig()
    let certificates: Array<X509Certificate> = X509Certificate.decodeFromPem(pem)
    let certificates_arr: Array<Certificate> = Array<Certificate>(certificates.size, { i => certificates[i]})
    cfg.verifyMode = CustomCA(certificates_arr)
    cfg.supportedAlpnProtocols = ["h2"]
    var con = dial("127.0.0.1:${port}", cfg: cfg)//linux
    let stream = con.newStreamClient()

    con.invoke("/helloworld.Greeter/SayHello1",req,resp)
    println(resp)
    stream.doStreamRequest("/helloworld.Greeter/SayHello")
    
    stream.doStreamRequest("/helloworld.Greeter/SayHello")
    //接收线程
    let fut: Future<String> = spawn { =>
        var resp = HelloReply()
        while(stream.recv(resp)) {
            println(resp.message)
        }
        return "finish"
    }

    for (_ in 0..num) {
        stream.sendStreamMsg(req)
    }
    stream.closeSend()//结束流

    let res: Option<String> = fut.get(1000 * 1000 * 1000 * 1000 )
    match (res) {
        case Some(v) => println(v)
        case None => throw Exception("gprc request timeout")
    }

    con.invoke("/helloworld.Greeter/SayHello1",req,resp)
    println(resp)
    return 0
}

//protobuf 依赖
public class HelloRequest <: MessageLite & TypedMessage<HelloRequest> {
    private var p_name: String

    public init(
        name!: String = ""
    ) {
        this.p_name = name
        m_init()
    }

    public init(src: HelloRequest__Builder) {
        this.p_name = src.name
        m_init()
    }

    private func m_init() {
        m_innerInit()
    }

    public mut prop name: String { get() { p_name } set(_v) { p_name = _v; markDirty() } }

    protected func m_size(): Int64 {
        var _x = 0
        if (!p_name.isEmpty()) { _x += 1 + binSize(p_name) }
        return _x
    }

    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        if (!p_name.isEmpty()) { _x.write(p_name.hashCode()) }
        return _x.finish()
    }

    protected func m_isEmpty(): Bool {
        p_name.isEmpty()
    }

    protected func m_clear(): Unit {
        p_name = ""
    }

    public operator func !=(src: HelloRequest): Bool { !(this == src) }
    public operator func ==(src: HelloRequest): Bool {
        hashCode() == src.hashCode() && size == src.size && unknownFields == src.unknownFields &&
        p_name == src.p_name
    }

    public func mergeFrom(src: HelloRequest): Unit {
        if (!src.p_name.isEmpty()) { p_name = src.p_name }
        if (!src.unknownFields.isEmpty()) { unknownFields.appendAll(src.unknownFields) }
        markDirty()
    }

    public func copyFrom(src: HelloRequest): HelloRequest {
        p_name = src.p_name
        m_innerCopyFrom(src)
        return this
    }

    public func clone(): HelloRequest {
        return HelloRequest().copyFrom(this)
    }

    public func unpack<T>(src: T): Unit where T <: BytesReader {
        while (!src.isEmpty()) {
            let _x = src.parseTag()
            match (_x[0]) {
                case 1 => p_name = src.parseString()
                case _ => m_unknown(src, _x)
            }
        }
        markDirty()
    }

    public func pack<T>(out: T): Unit where T <: BytesWriter {
        if (!p_name.isEmpty()) { out.append(10); out.packBin(p_name) }
        m_unknown(out)
    }

    public func toString(): String {
        var _x = StructPrinter()
        if (!p_name.isEmpty()) { _x.append("name", p_name) }
        return _x.done().toString()
    }

    public func packi(out: BytesWriter) { match (out) { case _x: SimpleWriter => pack(_x) case _ => pack(out) } }
    public func unpacki(src: BytesReader) { match (src) { case _x: SimpleReader => unpack(_x) case _ => unpack(src) } }

    public static func empty() { HelloRequest() }
    public static func newBuilder() { HelloRequest__Builder() }
    public static func fromBytes(src: Collection<Byte>) { let _x = HelloRequest(); _x.unpack(src); _x }
    public static func fromBytes(src: BytesReader) { let _x = HelloRequest(); _x.unpack(src); _x }
}

public class HelloRequest__Builder {
    public var name: String = ""
    public func setName(name: String) { this.name = name; return this }

    public func build() { HelloRequest(this) }
}

public class HelloReply <: MessageLite & TypedMessage<HelloReply> {
    private var p_message: String

    public init(
        message!: String = ""
    ) {
        this.p_message = message
        m_init()
    }

    public init(src: HelloReply__Builder) {
        this.p_message = src.message
        m_init()
    }

    private func m_init() {
        m_innerInit()
    }

    public mut prop message: String { get() { p_message } set(_v) { p_message = _v; markDirty() } }

    protected func m_size(): Int64 {
        var _x = 0
        if (!p_message.isEmpty()) { _x += 1 + binSize(p_message) }
        return _x
    }

    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        if (!p_message.isEmpty()) { _x.write(p_message.hashCode()) }
        return _x.finish()
    }

    protected func m_isEmpty(): Bool {
        p_message.isEmpty()
    }

    protected func m_clear(): Unit {
        p_message = ""
    }

    public operator func !=(src: HelloReply): Bool { !(this == src) }
    public operator func ==(src: HelloReply): Bool {
        hashCode() == src.hashCode() && size == src.size && unknownFields == src.unknownFields &&
        p_message == src.p_message
    }

    public func mergeFrom(src: HelloReply): Unit {
        if (!src.p_message.isEmpty()) { p_message = src.p_message }
        if (!src.unknownFields.isEmpty()) { unknownFields.appendAll(src.unknownFields) }
        markDirty()
    }

    public func copyFrom(src: HelloReply): HelloReply {
        p_message = src.p_message
        m_innerCopyFrom(src)
        return this
    }

    public func clone(): HelloReply {
        return HelloReply().copyFrom(this)
    }

    public func unpack<T>(src: T): Unit where T <: BytesReader {
        while (!src.isEmpty()) {
            let _x = src.parseTag()
            match (_x[0]) {
                case 1 => p_message = src.parseString()
                case _ => m_unknown(src, _x)
            }
        }
        markDirty()
    }

    public func pack<T>(out: T): Unit where T <: BytesWriter {
        if (!p_message.isEmpty()) { out.append(10); out.packBin(p_message) }
        m_unknown(out)
    }

    public func toString(): String {
        var _x = StructPrinter()
        if (!p_message.isEmpty()) { _x.append("message", p_message) }
        return _x.done().toString()
    }

    public func packi(out: BytesWriter) { match (out) { case _x: SimpleWriter => pack(_x) case _ => pack(out) } }
    public func unpacki(src: BytesReader) { match (src) { case _x: SimpleReader => unpack(_x) case _ => unpack(src) } }

    public static func empty() { HelloReply() }
    public static func newBuilder() { HelloReply__Builder() }
    public static func fromBytes(src: Collection<Byte>) { let _x = HelloReply(); _x.unpack(src); _x }
    public static func fromBytes(src: BytesReader) { let _x = HelloReply(); _x.unpack(src); _x }
}

public class HelloReply__Builder {
    public var message: String = ""
    public func setMessage(message: String) { this.message = message; return this }

    public func build() { HelloReply(this) }
}


//server端
public abstract class GreeterServer <: MessageLite {
	public func SayHello(hr: HelloRequest): (HelloReply, GrpcError)
	public func mustEmbedUnimplementedGreeterServer(): Unit

    protected func m_size(): Int64 { return 0 }
    protected func m_isEmpty(): Bool { return false }
    protected func m_clear(): Unit {}
    public func unpacki(_: BytesReader): Unit {}
    public func packi(_: BytesWriter): Unit {}
    public func toString(): String { return "" }
    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        return _x.finish()
    }

}

class UnimplementedGreeterServer <: GreeterServer {
	public func SayHello(_: HelloRequest): (HelloReply, GrpcError) {
		return (HelloReply(), GrpcError(Unimplemented, "method SayHello not implemented"))
	}
	public func mustEmbedUnimplementedGreeterServer() {}
}

interface UnsafeGreeterServer {
	func mustEmbedUnimplementedGreeterServer(): Unit
}

func Greeter_SayHello_Handler(_: Context, srv: MessageLite, dec: (MessageLite) -> GrpcError, interceptor: ?UnaryServerInterceptor): (MessageLite, GrpcError) {
	let inhr: HelloRequest = HelloRequest()

	var err: GrpcError = dec(inhr)
	if (!err.isNull()) {
		return (inhr, err)
	}
	match(interceptor){
		case Some(v) =>
				let info: UnaryServerInfo = UnaryServerInfo(
					server:     srv,
					fullMethod: "/helloworld.Greeter/SayHello1"
				)
				println("inhr.name = ${inhr.name}")
				inhr.name = "dec(inhr)解析到的HelloRequest是空的？？？-----000"
				func handlercs(req: Any): (Any, GrpcError) {
					return (srv as (GreeterServer)).getOrThrow().SayHello((req as HelloRequest).getOrThrow())
				}
				return v.f(inhr, info, UnaryHandler(handlercs))
		case None =>
				println("inhr.name = ${inhr.name}")
				//inhr.name = "dec(inhr)解析到的HelloRequest是空的？？？-----111"
				return (srv as (GreeterServer)).getOrThrow().SayHello(inhr)
				//return srv.SayHello(inhr)
	}
}

public func defaultAnyAndServerStreamToGrpcError(any: Any, ss: ServerStream): GrpcError {
    if (any is HelloRequest) {
        let hy: HelloRequest = (any as HelloRequest).getOrThrow()
        println("hy.name = ${hy.name}")
    } else if (any is HelloReply) {
		(any as HelloReply).getOrThrow()
		println("hy.message = hy.message")
	}

    var arr = ArrayList<String>()
	var req = HelloRequest()
	for (_ in 0..num) {
		ss.recvMsg(req)
        arr.append(req.name)
		println("req = ${req}")
	}
	let hr: HelloReply = HelloReply()
	for (i in 0..num) {
		hr.message = "ceshi-stream-" + i.toString() + ": "+arr[i]
		ss.sendMsg(hr)
	}


	return NULL_ERR
}

var Greeter_ServiceDesc = ServiceDesc(
	serviceName: "helloworld.Greeter",
	handlerType: -1,
	methods: [MethodDesc(methodName: "SayHello1", handler:  Greeter_SayHello_Handler)],
	streams: [StreamDesc(streamName: "SayHello", handler:  StreamHandler(defaultAnyAndServerStreamToGrpcError) )],
	metadata: "./helloworld.proto"
)

func registerGreeterServer(s: Server, srv: MessageLite) {
	s.registerService(Greeter_ServiceDesc, srv)
}

执行结果如下：

ss=127.0.0.1:46327
resp = {message: "World-0"}
resp = {message: "World-1"}
resp = {message: "World-2"}
resp = {message: "World-3"}
resp = {message: "World-4"}

压缩功能示例

可以启用默认 gzip 算法压缩请求数据，该示例需要和 test/LLT/grpc/compressor/helloworld.pb.cj 一起编译

示例代码如下：

public import std.net.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import std.collection.*
public import protobuf.*
public import std.sync.*
public import std.time.*


let strServer: String = "test"
class ServerTest <: GreeterServer {
    public func SayHello(_: HelloRequest): (HelloReply, GrpcError) {
        let hy: HelloReply = HelloReply()
        hy.message = strServer
        return (hy, GrpcError())
    }
    public func mustEmbedUnimplementedGreeterServer() {}
}

var port: UInt16 = 0
main() {

    let ss: TcpServerSocket = TcpServerSocket(bindAt: port)
    ss.bind()
    port = (ss.localAddress as IPSocketAddress).getOrThrow().port

    spawn{ =>
        let lis: GRPCServerSocket = GRPCServerSocket(GrpcSocketS(ss))
        let server: Server = Server()
        registerGreeterServer(server, ServerTest())
        server.serve(lis)
    }

    sleep(Duration.second)

    var req = HelloRequest()
    var resp = HelloReply()
    req.name = "World"

    var con = dial("127.0.0.1:${port}")//linux
    con.useCompressor()
    con.invoke("/helloworld.Greeter/SayHello",req,resp)
    println(resp)
    return 0
}

执行结果如下：

inhr.name = World
{message: "test pass"}

拦截器功能示例

拦截器提供对请求的参数和获取到的请求的拦截以实现对数据的检查和控制，一元请求拦截器示例如下，该示例需要和 test/LLT/grpc/interceptor/helloworld.pb.cj 一起编译

示例代码如下：

public import std.net.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import std.collection.*
public import protobuf.*
public import std.sync.*
public import std.time.*


let strServer: String = "test"
class ServerTest <: GreeterServer {
	public func SayHello(_: HelloRequest): (HelloReply, GrpcError) {
		let hy: HelloReply = HelloReply()
		hy.message = strServer
		return (hy, GrpcError())
	}
	public func mustEmbedUnimplementedGreeterServer() {}
}

var port: UInt16 = 0
main() {

	let ss: TcpServerSocket = TcpServerSocket(bindAt: port)
    ss.bind()
	port = (ss.localAddress as IPSocketAddress).getOrThrow().port

	spawn{ =>
        let lis: GRPCServerSocket = GRPCServerSocket(GrpcSocketS(ss))
        let server: Server = Server()
        registerGreeterServer(server, ServerTest())
        server.serve(lis)
	}

	sleep(Duration.second)

	var req = HelloRequest()
    req.name = "World"

    var opts = ArrayList<DialOption>()
    opts.append(withStreamInterceptor(unary()))

    var con = dial("127.0.0.1:${port}",opts)//linux

    let stream = con.newStreamClient()
    stream.doStreamRequest("/helloworld.Greeter/SayHello")
    let fut: Future<String> = spawn { =>
        var resp = HelloReply()
        while(stream.recv(resp)) {
            println(resp.message)
        }
        return "finish"
    }
    for (i in 0..3) {
        req.name = "World${i}"
        stream.sendStreamMsg(req)
    }
    stream.closeSend()

    let res: Option<String> = fut.get(Duration.nanosecond  * 1000 * 1000 * 1000 * 1000 )
    match (res) {
        case Some(v) => println(v)
        case None => throw Exception("gprc request timeout")
    }
    return 0
}

class unary <: StreamClientInterceptor {
    public func orderStreamClientInterceptor(method: String): Unit {
        if (method == "/helloworld.Greeter/SayHello") {
            println(method)
            println("method ok")
        } else {
            throw Exception("error method")
        }
    }

    public func orderStreamClientInterceptorReq(req: MessageLite): Unit {
        var r = (req as HelloRequest).getOrThrow()
        if (r.name.size > 1) {
            println(r.name)
            println("request ok")
        }
        else {
            throw Exception("error request")
        }  
    }

    public func orderStreamClientInterceptorResp(resp: MessageLite): Unit {
        var r = (resp as HelloReply).getOrThrow()
        if (r.message == "test") {
            println("response ok")
        }
        else {
            throw Exception("error response")
        }  
    }
}

执行结果如下：

inhr.name = World
测试服务端一元拦截器调用！
/helloworld.Greeter/SayHello
method ok
World
request ok

域名解析功能示例

客户端可以启用内置域名解析功能，提供对服务器域名的解析并连接，示例如下:

示例代码如下：

public import std.net.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import std.collection.*
public import rpc4cj.resolver.*
public import rpc4cj.resolver.dns.*
public import std.sync.*
public import std.time.*


var port: UInt16 = 0
main() {
    let ss: TcpServerSocket = TcpServerSocket(bindAt: port)
	ss.bind()
	port = (ss.localAddress as IPSocketAddress).getOrThrow().port
    spawn{ =>
        let lis: GRPCServerSocket = GRPCServerSocket(GrpcSocketS(ss))
        let server: Server = Server()
        server.serve(lis)
    }
    sleep(Duration.second * 3)
    try{
        dial("passthrough:///127.0.0.1:${port}")//reslove localhost with
    } catch (e: Exception) {
          println("reslove failed")
          return 1
    }

    return 0
}

执行结果如下：

reslove pass

TLS 安全连接功能示例

TLS 是一种网络传输层安全协议，可以为 rpc4cj 客户端应用程序和服务器端应用程序之间提供安全连接，示例如下:
注：示例中使用到的密钥仅供参考，用户需要自行准备。

示例代码如下：

// DEPENDENCE: ./example-cert.pem
// DEPENDENCE: ./example-key.pem
// EXEC: cjc %import-path %L %l %f
// EXEC: ./main

public import std.net.*
public import std.io.*
public import std.collection.*
public import stdx.net.tls.*
public import stdx.net.tls.common.*
public import crypto.x509.*
public import stdx.crypto.keys.*
public import stdx.crypto.common.*
public import rpc4cj.grpc.*
public import rpc4cj.transport.*
public import rpc4cj.exceptions.*
public import rpc4cj.util.*
public import protobuf.*
public import std.sync.*
public import std.time.*
public import std.fs.*


let strServer: String = "test"
var port: UInt16 = 0
let num: Int64 = 5

main() {
    let pem = String.fromUtf8(readToEnd(File("./example-cert.pem", OpenOption.Open(true, false))))
    let key = String.fromUtf8(readToEnd(File("./example-key.pem", OpenOption.Open(true, false))))
    var cfg_temp = TlsServerConfig(X509Certificate.decodeFromPem(pem),GeneralPrivateKey.decodeFromPem(key))
    cfg_temp.supportedAlpnProtocols = ["h2"]
    let ss: TcpServerSocket = TcpServerSocket(bindAt: port)
    ss.bind()
    let cfgs = cfg_temp
    port = (ss.localAddress as IPSocketAddress).getOrThrow().port
    
    spawn{ =>
        let serve = TlsSocket.server(ss.accept(),serverConfig: cfgs)
        let lis: GRPCServerSocket = GRPCServerSocket(GrpcTlsS(serve))
        let server: Server = Server()
        registerGreeterServer(server, ServerTest())
        server.serve(lis)
    }

    sleep(Duration.second)

    var req = HelloRequest()
    req.name = "World"
    var cfg = TlsClientConfig()
    let certificates: Array<X509Certificate> = X509Certificate.decodeFromPem(pem)
    let certificates_arr: Array<Certificate> = Array<Certificate>(certificates.size, { i => certificates[i]})
    cfg.verifyMode = CustomCA(certificates_arr)
    cfg.supportedAlpnProtocols = ["h2"]
    let con = dial("127.0.0.1:${port}", cfg: cfg)//linux
    let stream = con.newStreamClient()



    stream.doStreamRequest("/helloworld.Greeter/SayHello")

    let fut: Future<String> = spawn { =>
        var recv = HelloReply()
        while(stream.recv(recv)) {
            println(recv.message)
        }
        return "finish"
    }
    
    for (_ in 0..num) {
        stream.sendStreamMsg(req)
    }

    stream.closeSend()

    let res: Option<String> = fut.get(Duration.nanosecond  * 1000 * 1000 * 1000 * 10 )
    match (res) {
        case Some(v) => println(v)
        case None => throw Exception("gprc request timeout")
    }
    return 0
}


//protobuf 依赖
class ServerTest <: GreeterServer {
    public func SayHello(hr: HelloRequest): (HelloReply, GrpcError) {
        let hy: HelloReply = HelloReply()
        hy.message = "ceshi-20221227"
        println("received: hr.name = ${hr.name}")
        return (hy, GrpcError())
    }
    public func mustEmbedUnimplementedGreeterServer() {}
}

public class HelloRequest <: MessageLite & TypedMessage<HelloRequest> {
    private var p_name: String

    public init(
        name!: String = ""
    ) {
        this.p_name = name
        m_init()
    }

    public init(src: HelloRequest__Builder) {
        this.p_name = src.name
        m_init()
    }

    private func m_init() {
        m_innerInit()
    }

    public mut prop name: String { get() { p_name } set(_v) { p_name = _v; markDirty() } }

    protected func m_size(): Int64 {
        var _x = 0
        if (!p_name.isEmpty()) { _x += 1 + binSize(p_name) }
        return _x
    }

    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        if (!p_name.isEmpty()) { _x.write(p_name.hashCode()) }
        return _x.finish()
    }

    protected func m_isEmpty(): Bool {
        p_name.isEmpty()
    }

    protected func m_clear(): Unit {
        p_name = ""
    }

    public operator func !=(src: HelloRequest): Bool { !(this == src) }
    public operator func ==(src: HelloRequest): Bool {
        hashCode() == src.hashCode() && size == src.size && unknownFields == src.unknownFields &&
        p_name == src.p_name
    }

    public func mergeFrom(src: HelloRequest): Unit {
        if (!src.p_name.isEmpty()) { p_name = src.p_name }
        if (!src.unknownFields.isEmpty()) { unknownFields.appendAll(src.unknownFields) }
        markDirty()
    }

    public func copyFrom(src: HelloRequest): HelloRequest {
        p_name = src.p_name
        m_innerCopyFrom(src)
        return this
    }

    public func clone(): HelloRequest {
        return HelloRequest().copyFrom(this)
    }

    public func unpack<T>(src: T): Unit where T <: BytesReader {
        while (!src.isEmpty()) {
            let _x = src.parseTag()
            match (_x[0]) {
                case 1 => p_name = src.parseString()
                case _ => m_unknown(src, _x)
            }
        }
        markDirty()
    }

    public func pack<T>(out: T): Unit where T <: BytesWriter {
        if (!p_name.isEmpty()) { out.append(10); out.packBin(p_name) }
        m_unknown(out)
    }

    public func toString(): String {
        var _x = StructPrinter()
        if (!p_name.isEmpty()) { _x.append("name", p_name) }
        return _x.done().toString()
    }

    public func packi(out: BytesWriter) { match (out) { case _x: SimpleWriter => pack(_x) case _ => pack(out) } }
    public func unpacki(src: BytesReader) { match (src) { case _x: SimpleReader => unpack(_x) case _ => unpack(src) } }

    public static func empty() { HelloRequest() }
    public static func newBuilder() { HelloRequest__Builder() }
    public static func fromBytes(src: Collection<Byte>) { let _x = HelloRequest(); _x.unpack(src); _x }
    public static func fromBytes(src: BytesReader) { let _x = HelloRequest(); _x.unpack(src); _x }
}

public class HelloRequest__Builder {
    public var name: String = ""
    public func setName(name: String) { this.name = name; return this }

    public func build() { HelloRequest(this) }
}

public class HelloReply <: MessageLite & TypedMessage<HelloReply> {
    private var p_message: String

    public init(
        message!: String = ""
    ) {
        this.p_message = message
        m_init()
    }

    public init(src: HelloReply__Builder) {
        this.p_message = src.message
        m_init()
    }

    private func m_init() {
        m_innerInit()
    }

    public mut prop message: String { get() { p_message } set(_v) { p_message = _v; markDirty() } }

    protected func m_size(): Int64 {
        var _x = 0
        if (!p_message.isEmpty()) { _x += 1 + binSize(p_message) }
        return _x
    }

    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        if (!p_message.isEmpty()) { _x.write(p_message.hashCode()) }
        return _x.finish()
    }

    protected func m_isEmpty(): Bool {
        p_message.isEmpty()
    }

    protected func m_clear(): Unit {
        p_message = ""
    }

    public operator func !=(src: HelloReply): Bool { !(this == src) }
    public operator func ==(src: HelloReply): Bool {
        hashCode() == src.hashCode() && size == src.size && unknownFields == src.unknownFields &&
        p_message == src.p_message
    }

    public func mergeFrom(src: HelloReply): Unit {
        if (!src.p_message.isEmpty()) { p_message = src.p_message }
        if (!src.unknownFields.isEmpty()) { unknownFields.appendAll(src.unknownFields) }
        markDirty()
    }

    public func copyFrom(src: HelloReply): HelloReply {
        p_message = src.p_message
        m_innerCopyFrom(src)
        return this
    }

    public func clone(): HelloReply {
        return HelloReply().copyFrom(this)
    }

    public func unpack<T>(src: T): Unit where T <: BytesReader {
        while (!src.isEmpty()) {
            let _x = src.parseTag()
            match (_x[0]) {
                case 1 => p_message = src.parseString()
                case _ => m_unknown(src, _x)
            }
        }
        markDirty()
    }

    public func pack<T>(out: T): Unit where T <: BytesWriter {
        if (!p_message.isEmpty()) { out.append(10); out.packBin(p_message) }
        m_unknown(out)
    }

    public func toString(): String {
        var _x = StructPrinter()
        if (!p_message.isEmpty()) { _x.append("message", p_message) }
        return _x.done().toString()
    }

    public func packi(out: BytesWriter) { match (out) { case _x: SimpleWriter => pack(_x) case _ => pack(out) } }
    public func unpacki(src: BytesReader) { match (src) { case _x: SimpleReader => unpack(_x) case _ => unpack(src) } }

    public static func empty() { HelloReply() }
    public static func newBuilder() { HelloReply__Builder() }
    public static func fromBytes(src: Collection<Byte>) { let _x = HelloReply(); _x.unpack(src); _x }
    public static func fromBytes(src: BytesReader) { let _x = HelloReply(); _x.unpack(src); _x }
}

public class HelloReply__Builder {
    public var message: String = ""
    public func setMessage(message: String) { this.message = message; return this }

    public func build() { HelloReply(this) }
}

//server 端
public abstract class GreeterServer <: MessageLite {
    public func SayHello(hr: HelloRequest): (HelloReply, GrpcError)
    public func mustEmbedUnimplementedGreeterServer(): Unit

    protected func m_size(): Int64 { return 0 }
    protected func m_isEmpty(): Bool { return false }
    protected func m_clear(): Unit {}
    public func unpacki(_: BytesReader): Unit {}
    public func packi(_: BytesWriter): Unit {}
    public func toString(): String { return "" }
    protected func m_hashCode(): Int64 {
        var _x = DefaultHasher()
        return _x.finish()
    }
}

class UnimplementedGreeterServer <: GreeterServer {
    public func SayHello(_: HelloRequest): (HelloReply, GrpcError) {
        return (HelloReply(), GrpcError(Unimplemented, "method SayHello not implemented"))
    }
    public func mustEmbedUnimplementedGreeterServer() {}
}

interface UnsafeGreeterServer {
    func mustEmbedUnimplementedGreeterServer(): Unit
}

public func defaultAnyAndServerStreamToGrpcError(any: Any, ss: ServerStream): GrpcError {
    if (any is HelloRequest) {
        let hy: HelloRequest = (any as HelloRequest).getOrThrow()
        println("hy.name = ${hy.name}")
    } else if (any is HelloReply) {
        let hy: HelloReply = (any as HelloReply).getOrThrow()
        println("hy.message = ${hy.message}")
    }

    let hr: HelloRequest = HelloRequest()
    for (i in 0..num) {
        hr.name = "ceshi-20221227-stream-" + i.toString()
        ss.sendMsg(hr)
    }
    
    var resp = HelloReply()
    for (_ in 0..num) {
        ss.recvMsg(resp)
        println("resp = ${resp}")
    }

    return NULL_ERR
}

var Greeter_ServiceDesc = ServiceDesc(
    serviceName: "helloworld.Greeter",
    handlerType: -1,
    streams: [StreamDesc(streamName: "SayHello", handler:  StreamHandler(defaultAnyAndServerStreamToGrpcError) )],
    metadata: "./helloworld.proto"
)

func registerGreeterServer(s: Server, srv: MessageLite) {
    s.registerService(Greeter_ServiceDesc, srv)
}

执行结果如下：

ss=127.0.0.1:50051
inhr.name = World
测试server端-20220919
received: hr.name = World
result: {message: "测试server端-20230224......哈哈哈哈，测试通了！"}
result: {"message": "Hello World"}

开源协议

本项目基于 Apache License 2.0 License，请自由地享受和参与开源。

参与贡献

欢迎给我们提交PR，欢迎给我们提交Issue，欢迎参与任何形式的贡献。