1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
func main() {
	h, err := InitServerEnv()
	if err != nil {
		log.Fatal(err)
		return
	}
	pb.RegisterGreeterServer(h, &s2s{})
	if err := rpc.Serve(h); err != nil {
		log.Fatalf("failed to serve: %v", err)
	}
}

type s2s struct{}

func (s *s2s) SayHello(ctx context.Context, helloRequest *pb.HelloRequest) (*pb.HelloReply, error) {
	log.Info(helloRequest)
	return &pb.HelloReply{Message: "HelloReplyContent"}, nil
}

func (s *s2s) SayHello2(ctx context.Context, helloRequest *pb.HelloRequest2) (*pb.HelloReply2, error) {
	log.Info(helloRequest)
	return &pb.HelloReply2{ReplyNum:helloRequest.Num, Res: true}, nil
}

1
2
3
4
type Package struct {
	ServiceName string
	Data []byte
}

1
2
3
4
5
type Transport interface {
	Setup(network string, address string) error
	Recv() (*Package, error)
	Send(pak *Package) error
}

1
2
3
4
type Codec interface {
	Encode(interface{}) ([]byte, error)
	Decode([]byte, interface{}) error
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
type Service interface {
	Register(serviceDesc interface{}, serviceImpl interface{}) error
	Serve() error
	Close(chan struct{}) error
}

type ServiceHandler interface {
	Name() string
	HandleRPC(context.Context, string, Codec, *transport.Package) ([]byte, error)
}

type CustomService struct {
	requestId      int64
	ctx            context.Context
	cancel         context.CancelFunc
	trans          transport.Transport
	serviceHandler ServiceHandler
	d              Codec
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func (h *CustomService) internalRecv() (int, error) {
    pak, err := h.trans.Recv()
    ...
    go func(pak *transport.Package) {
        ...
        h.internalHandle(pak)
    }(pak)
}

func (h *CustomService) internalHandle(pak *transport.Package) {
	var param []interface{}
	var rpcName string
	err := h.d.Decode(pak.Data, &param)
	rpcName = param[0].(string)
    ...
	rspBin, err := h.handleRPC(rpcName, pak)
	if err == nil && len(rspBin) > 0 {
		var sendPak = *pak
		sendPak.Data = rspBin
		err = h.Send(&sendPak)
	}
}

func (h *CustomService) Send(pak *transport.Package) error {
	return h.trans.Send(pak)
}

1
2
3
4
5
6
7
8
9
10
func (h *greeterHandler) HandleRPC(ctx context.Context, rpcName string, d rpc.Codec, pak *transport.Package) ([]byte, error) {
	value := reflect.ValueOf(h)
	args := []reflect.Value{reflect.ValueOf(ctx), reflect.ValueOf(rpcName), reflect.ValueOf(d), reflect.ValueOf(pak)}
	f := value.MethodByName(rpcName)
	res := f.Call(args)
	if res[0].IsNil() {
		return nil, res[1].Interface().(error)
	}
	return res[0].Bytes(), nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
func (h *greeterHandler) SayHello(ctx context.Context, rpcName string, d rpc.Codec, pak *transport.Package) ([]byte, error) {
	var helloRequest HelloRequest
	var inVarList []interface{}
	var temp string
	var callback string
	inVarList = append(inVarList, &temp)
	inVarList = append(inVarList, &callback)
	inVarList = append(inVarList, &helloRequest.Name)
	err := rpc.DecodeArchiverWithTrace(rpcName, d, pak, inVarList...)
	if err != nil {
		return nil, err
	}
	helloReply, err := h.GreeterCustomServer.SayHello(ctx, &helloRequest)
	if err != nil {
		return nil, fmt.Errorf(`rpc failed in [%s]: %s`, rpcName, err.Error())
	}
	var outVarList []interface{}
	outVarList = append(outVarList, &callback)
	outVarList = append(outVarList, &helloReply.Message)
	return d.Encode(outVarList)
}

func (g *custom) generateServerMethod(servName string, method *pb.MethodDescriptorProto, messages []*descriptorpb.DescriptorProto) string {
	methName := generator.CamelCase(method.GetName())
	hname := methName
	serveType := servName + "Handler"
	inType := g.typeName(method.GetInputType())
	outType := g.typeName(method.GetOutputType())
	inputPb := g.getPbByName(messages, inType)
	outPb := g.getPbByName(messages, outType)
	if !method.GetClientStreaming() && !method.GetServerStreaming() {
		g.P("func (h *", unexport(servName), "Handler) ", hname, "(ctx ", contextPkg, ".Context, rpcName string, d rpc.Codec, pak *", transportPkg, ".Package) ([]byte, error) {")
		g.P("var ", unexport(inType), " ", inType)
		g.P("var inVarList []interface{}")
		g.P("var temp string")
		g.P("var callback string")
		g.P("inVarList = append(inVarList, &temp)")
		g.P("inVarList = append(inVarList, &callback)")
		// ...
		for _, field := range inputPb.GetField() {
			g.P("inVarList = append(inVarList, &", unexport(inType), ".", generator.CamelCase(field.GetName()), ")")
		}
		g.P("err := rpc.DecodeArchiverWithTrace(rpcName, d, pak, inVarList...)")
		g.P("if err != nil {")
		g.P("return nil, err")
		g.P("}")
		g.P(unexport(outType), ", err := h.", servName, "CustomServer.", methName, "(ctx, &", unexport(inType),")")
		g.P("if err != nil {")
		g.P("return nil, fmt.Errorf(`rpc failed in [%s]: %s`, rpcName, err.Error())")
		g.P("}")
		g.P("var outVarList []interface{}")
		g.P("outVarList = append(outVarList, &callback)")
		// ...
		for _, field := range outPb.GetField() {
			g.P("outVarList = append(outVarList, &", unexport(outType), ".", generator.CamelCase(field.GetName()), ")")
		}
		g.P("return d.Encode(outVarList)")
		g.P("}")
		g.P()
		return hname
	}
    ...
}