ypgrpc-go-app and gRPC Services
This tutorial provides a basic Go introduction to working with ypgrpc-go-app and gRPC callbacks.
The first step is to define the gRPC Service and the method
request and response types using protocol buffers. For the
complete .proto file, see
$HOME/go/src/ypgrpc/examples/example/example.proto
gRPC defines five kinds of service methods, all of which are used in the ExampleService service:
An empty request and empty response RPC
/* Empty request And Empty response RPC */ rpc EmptyCall(google.protobuf.Empty) returns (google.protobuf.Empty);
A simple RPC where the client sends a request to the server using the stub and waits for a response to come back, just like a normal function call.
/* RPC that represent single request and response * The server returns the client payload as-is. */ rpc UnaryCall(SimpleRequest) returns (SimpleResponse);
A server-side streaming RPC where the client sends a request to the server and gets a stream to read a sequence of messages back. The client reads from the returned stream until there are no more messages. To specify a server-side streaming method, place the stream keyword before the response type.
/* RPC that represent single request and a streaming response * The server returns the payload with client desired type and sizes. */ rpc StreamingOutputCall(StreamingOutputCallRequest) returns (stream StreamingOutputCallResponse);
A client-side streaming RPC where the client writes a sequence of messages and sends them to the server, again using a provided stream. Once the client has finished writing the messages, it waits for the server to read them all and return its response. To specify a client-side streaming method, place the stream keyword before the request type.
/* RPC that represent a sequence of requests and a sngle responses * The server returns the aggregated size of client payload as the result. */ rpc StreamingInputCall(stream StreamingInputCallRequest) returns (StreamingInputCallResponse);
A bidirectional streaming RPC where both sides send a sequence of messages using a read-write stream. The two streams operate independently, so clients and servers can read and write in whatever order they like: for example, the server could wait to receive all the client messages before writing its responses, or it could alternately read a message then write a message, or some other combination of reads and writes. The order of messages in each stream is preserved. To specify this type of method, place the stream keyword before both the request and the response.
/* RPC that represent a sequence of requests and responses * with each request served by the server immediately. * As one request could lead to multiple responses, this interface * demonstrates the idea of full duplexing. */ rpc FullDuplexCall(stream StreamingOutputCallRequest) returns (stream StreamingOutputCallResponse); /* RPC that represent a sequence of requests and responses. * The server buffers all the client requests and then serves them in order. * A stream of responses are returned to the client when the server starts with * first request. */ rpc HalfDuplexCall(stream StreamingOutputCallRequest) returns (stream StreamingOutputCallResponse);
The .proto file also contains protocol buffer message type definitions for all the request and response types used in our service methods - for example, here is the SimpleRequest message type:
/* Unary request */
message SimpleRequest {
EchoStatus response_status = 1;
User user = 2;
}
ypgrpc-go-app Application
The ypgrpc-go-app application provides an interface to add new gRPC Services and implement new Methods.
This application shows an example of how the gRPC interface can be used and how the Services and methods can be implemented.
Example ypgrpc-go-app Application:
/** helloworldServer is used to implement helloworld.GreeterServer */
type helloworldServer struct {
helloworld.UnimplementedGreeterServer
}
/** exampleServer is used to implement example.ExampleServiceServer */
type exampleServer struct {
example.UnimplementedExampleServiceServer
}
/** @} */
/********************************************************************
* *
* F U N C T I O N S *
* *
*********************************************************************/
/* SayHello implements helloworld.GreeterServer */
func (s *helloworldServer) SayHello (ctx context.Context,
in *helloworld.HelloRequest) (
*helloworld.HelloReply,
error) {
log.Log_info("\n\nHelloRequest:")
log.Log_dump_structure(in)
return &helloworld.HelloReply{
Message: "Hello " + in.GetName(),
}, nil
}
/* SayHelloAgain implements helloworld.GreeterServer */
func (s *helloworldServer) SayHelloAgain (ctx context.Context,
in *helloworld.HelloRequest) (
*helloworld.HelloReply,
error) {
log.Log_info("\n\nHelloRequest:")
log.Log_dump_structure(in)
return &helloworld.HelloReply{
Message: "Say Hello Again" + in.GetName(),
}, nil
}
/* SayHelloOneMore implements helloworld.GreeterServer */
func (s *helloworldServer) SayHelloOneMore (ctx context.Context,
in *helloworld.HelloRequest) (
*helloworld.HelloReply,
error) {
log.Log_info("\n\nHelloRequest:")
log.Log_dump_structure(in)
return &helloworld.HelloReply{
Message: "Say Hello OneMore" + in.GetName(),
}, nil
}
/* Empty request And Empty response RPC */
func (exampleServer) EmptyCall (ctx context.Context,
in *emptypb.Empty) (
*emptypb.Empty,
error) {
log.Log_info("\n\nEmptyRequest:")
log.Log_dump_structure(in)
/* Empty input and putput.
* Run your Instrumentation here
*/
return in, nil
}
/* RPC that represent single request and response
* The server returns the client payload as-is.
*/
func (exampleServer) UnaryCall (ctx context.Context,
in *example.SimpleRequest) (
*example.SimpleResponse,
error) {
log.Log_info("\n\nSimpleRequest:")
log.Log_dump_structure(in)
/* The Incoming Code must be 0 to signal that request is well formed */
if in.ResponseStatus != nil && in.ResponseStatus.Code != int32(codes.OK) {
log.Log_info("\nReceived Error Code: %v",
in.GetResponseStatus().GetCode())
return nil, status.Error(codes.Code(in.ResponseStatus.Code), "error")
}
return &example.SimpleResponse{
User: &example.User{
Id: in.GetUser().GetId(),
Name: in.GetUser().GetName(),
},
}, nil
}
/* RPC that represent single request and a streaming response
* The server returns the payload with client desired type and sizes.
*/
func (exampleServer) StreamingOutputCall (in *example.StreamingOutputCallRequest,
stream example.ExampleService_StreamingOutputCallServer) error {
/* Example Request:
{ "response_parameters":[
{"size":10,"interval_us":5},
{"size":10,"interval_us":6}
],
"user": {"id":13,"name":"Example-1"},
"response_status":{"code":0,"message":"Example message 1"}
}
*/
log.Log_info("\n\nStreamingOutputCallRequest:")
log.Log_dump_structure(in)
rsp := &example.StreamingOutputCallResponse{
User: &example.User{},
}
/* Starting/ Closing a Server Stream. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"StreamingOutputCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"StreamingOutputCall")
count := int32(1)
for _, param := range in.ResponseParameters {
log.Log_info("\nInterval between responses: %v\n",
param.GetIntervalUs())
/* Wait as specified in the interval parameter */
time.Sleep(time.Duration(param.GetIntervalUs()) * time.Second)
if stream.Context().Err() != nil {
/* Closing a Server Stream. Update monitoring information. */
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += in.GetUser().GetName()
}
count++
rsp.User.Id = count
rsp.User.Name = buf
if err := stream.Send(rsp); err != nil {
/* Closing a Server Stream. Update monitoring information. */
return err
}
}
log.Log_info("\nDone Streaming")
return nil
}
/* RPC that represent a sequence of requests and a single responses
* The server returns the aggregated size of client payload as the result.
*/
func (exampleServer) StreamingInputCall (stream example.ExampleService_StreamingInputCallServer) error {
log.Log_info("\n\nExampleService_StreamingInputCallServer:")
log.Log_dump_structure(stream)
/* Starting/ Closing a Client Stream. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"StreamingInputCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"StreamingInputCall")
size := 0
for {
req, err := stream.Recv()
if err == io.EOF {
return stream.SendAndClose(&example.StreamingInputCallResponse{
AggregatedPayloadSize: int32(size),
})
}
size += len(req.User.Name)
if err != nil {
return err
}
}
}
/* RPC that represent a sequence of requests and responses
* with each request served by the server immediately.
* As one request could lead to multiple responses, this interface
* demonstrates the idea of full duplexing.
*/
func (exampleServer) FullDuplexCall (stream example.ExampleService_FullDuplexCallServer) error {
log.Log_info("\n\nExampleService_FullDuplexCallServer:")
log.Log_dump_structure(stream)
/* Starting/ Closing a Client/ Server Streams. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"FullDuplexCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"FullDuplexCall")
for {
req, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return status.Error(codes.Internal, err.Error())
}
if req.ResponseStatus != nil && req.ResponseStatus.Code != int32(codes.OK) {
return status.Error(codes.Code(req.ResponseStatus.Code), "error")
}
resp := &example.StreamingOutputCallResponse{User: &example.User{}}
for _, param := range req.ResponseParameters {
if stream.Context().Err() != nil {
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += req.GetUser().GetName()
}
resp.User.Name = buf
if err := stream.Send(resp); err != nil {
return err
}
}
}
}
/* RPC that represent a sequence of requests and responses.
* The server buffers all the client requests and then serves them in order.
* A stream of responses are returned to the client when the server starts with
* first request.
*/
func (exampleServer) HalfDuplexCall (stream example.ExampleService_HalfDuplexCallServer) error {
log.Log_info("\n\nExampleService_HalfDuplexCallServer:")
log.Log_debug_dump(stream)
/* Starting/ Closing a Client/ Server Streams. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"HalfDuplexCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"HalfDuplexCall")
requests := []*example.StreamingOutputCallRequest{}
for {
req, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return status.Error(codes.Internal, err.Error())
}
requests = append(requests, req)
}
for _, req := range requests {
resp := &example.StreamingOutputCallResponse{User: &example.User{}}
for _, param := range req.ResponseParameters {
if stream.Context().Err() != nil {
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += req.GetUser().GetName()
}
resp.User.Name = buf
if err := stream.Send(resp); err != nil {
return err
}
}
}
return nil
}
/**
* @brief MAIN IO server loop for the gRPC manager
*
*/
func main () {
var res error = nil
/* Connecton to netconfd-pro server:
* 1) Parse CLI parameters:
* - Host
* - subsys-id
* - user
* - proto files
* - etc
* 2) Open Socket and send NCX-Connect request
* 3) Start to listen on the socket (select loop)
* 4) Register YControl service (gRPC service)
* 5) send <capability-ad-event> event
* 7) Start to listen for any request <-> response
*
*/
/* Parse all the CLI parameters */
res = cli.ParseCLIParameters()
if res != nil {
utils.Check_error(res)
}
/* set logging parameters */
log.SetLevel()
log.SetLogOut()
/* Print all the CLi parameters provided */
log.Log_dump_structure(cli.GetCliOptions())
log.Log_info("\n\nStarting ypgrpc-go-app...")
log.Log_info("\nCopyright (c) 2021, YumaWorks, Inc., " +
"All Rights Reserved.\n")
/* Connect the the server */
conn, res := netconfd_connect.Connect_netconfd()
if res != nil {
utils.Check_error(res)
}
/* Defer is used to ensure that a function call is performed
* later in a program’s execution, usually for purposes of cleanup.
* defer is often used where e.g. ensure and finally would be used
* in other languages.
*/
defer conn.Close()
/* Get the actual TCP host address to use for gRPC server,
* Address CLI parameter + port number,
* By default this will be:
* 127.0.0.1:50830
*/
addr := netconfd_connect.GetServerAddr()
log.Log_info("\nStarting the gRPC server ...")
log.Log_info("\nListening for client request on '%s'",
addr)
/* All initialiation is done
* Start the gRPC server to listen for clients
*/
lis, err := net.Listen("tcp", addr)
if err != nil {
log.Log_error("\nError: failed to Listen (%v)", err)
}
/* Start the gRPC server and Register all the gRPC Services */
grpcServer := grpc.NewServer()
/* Register All the Services here */
helloworld.RegisterGreeterServer(grpcServer, &helloworldServer{})
example.RegisterExampleServiceServer(grpcServer, &exampleServer{})
log.Log_info("\nypgrpc_server: Starting to serve")
if err := grpcServer.Serve(lis); err != nil {
log.Log_error("\nError: failed to Serve (%v)", err)
}
result, res := ioutil.ReadAll(conn)
utils.Check_error(res)
log.Log_info("\n%s", string(result))
log.CloseLogOut()
os.Exit(0)
} /* main */
ypgrpc-go-app Interface Functions
The ypgrpc-go-app application is an YControl subsystem (similar to db-api-app) that has multiple API to communicate with the netconfd-pro server.
Setup
netconfd_connect.Connect_netconfd: Initialize the YP-gRPC service with the YControl subsystem and advertise gRPC server capabilities to the netconfd-pro
Update monitoring information
netconfd_connect.Open_streams: Sends subsystem event to advertise all the gRPC available and active capabilities during the registration time
netconfd_connect.Close_streams: Sends subsystem event to advertise that the gRPC server or client stream was closed
gRPC Service Implementation
The ypgrpc-go-app example gRPC server has a exampleServer structure type that implements the generated ExampleService interface:
type exampleServer struct {
}
func (exampleServer) EmptyCall (ctx context.Context,
in *emptypb.Empty) (
*emptypb.Empty,
error) {
}
func (exampleServer) UnaryCall (ctx context.Context,
in *example.SimpleRequest) (
*example.SimpleResponse,
error) {
}
func (exampleServer) StreamingOutputCall (in *example.StreamingOutputCallRequest,
stream example.ExampleService_StreamingOutputCallServer) error {
}
func (exampleServer) StreamingInputCall (stream example.ExampleService_StreamingInputCallServer) error {
}
func (exampleServer) FullDuplexCall (stream example.ExampleService_FullDuplexCallServer) error {
}
func (exampleServer) HalfDuplexCall (stream example.ExampleService_HalfDuplexCallServer) error {
}
Empty RPC
The exampleServer implements all the service methods.
/* Empty request And Empty response RPC */
func (exampleServer) EmptyCall (ctx context.Context,
in *emptypb.Empty) (
*emptypb.Empty,
error) {
log.Log_info("\n\nEmptyRequest:")
log.Log_dump_structure(in)
/* Empty input and output.
* Run your Instrumentation here
*/
return in, nil
}
Simple RPC
A simple UnaryCall, which just gets a SimpleRequest from the client and returns the corresponding SimpleResponse from the server.
The method is passed a context object for the RPC and the client’s SimpleRequest protocol buffer request. It returns a SimpleResponse protocol buffer object with the response information and an error.
In this method, the SimpleResponse is set with the appropriate information, and then returned with an nil error.
/* RPC that represent single request and response
* The server returns the client payload as-is.
*/
func (exampleServer) UnaryCall (ctx context.Context,
in *example.SimpleRequest) (
*example.SimpleResponse,
error) {
log.Log_info("\n\nSimpleRequest:")
log.Log_dump_structure(in)
/* The Incoming Code must be 0 to signal that request is well formed */
if in.ResponseStatus != nil && in.ResponseStatus.Code != int32(codes.OK) {
log.Log_info("\nReceived Error Code: %v",
in.GetResponseStatus().GetCode())
return nil, status.Error(codes.Code(in.ResponseStatus.Code), "error")
}
return &example.SimpleResponse{
User: &example.User{
Id: in.GetUser().GetId(),
Name: in.GetUser().GetName(),
},
}, nil
}
Server-side Streaming RPC
The StreamingOutputCall is a server-side streaming RPC, which sends multiple StreamingOutputCallResponse objects to the client.
/* RPC that represent single request and a streaming response
* The server returns the payload with client desired type and sizes.
*/
func (exampleServer) StreamingOutputCall (in *example.StreamingOutputCallRequest,
stream example.ExampleService_StreamingOutputCallServer) error {
/* Example Request:
{ "response_parameters":[
{"size":10,"interval_us":5},
{"size":10,"interval_us":6}
],
"user": {"id":13,"name":"Example-1"},
"response_status":{"code":0,"message":"Example message 1"}
}
*/
log.Log_info("\n\nStreamingOutputCallRequest:")
log.Log_dump_structure(in)
rsp := &example.StreamingOutputCallResponse{
User: &example.User{},
}
/* Starting/ Closing a Server Stream. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"StreamingOutputCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"StreamingOutputCall")
count := int32(1)
for _, param := range in.ResponseParameters {
log.Log_info("\nInterval between responses: %v\n",
param.GetIntervalUs())
/* Wait as specified in the interval parameter */
time.Sleep(time.Duration(param.GetIntervalUs()) * time.Second)
if stream.Context().Err() != nil {
/* Closing a Server Stream. Update monitoring information. */
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += in.GetUser().GetName()
}
count++
rsp.User.Id = count
rsp.User.Name = buf
if err := stream.Send(rsp); err != nil {
/* Closing a Server Stream. Update monitoring information. */
return err
}
}
log.Log_info("\nDone Streaming")
return nil
}
Instead of getting simple request and response objects in the method parameters, this time a request object and a special StreamingOutputCallRequest object is used to write the responses.
The server returns the aggregated size of client payload as the result.
In the method, as many StreamingOutputCallResponse objects are populated as needed, writing them to the ExampleService_StreamingOutputCallServer using its Send() method.
Finally, a nil error to indicate that the function is finished writing responses. Should any error happen in this call, the return a non-nil error; the gRPC layer will translate it into an appropriate RPC status to be sent to the client.
Client-side Streaming RPC
The client-side streaming method RecordRoute function is used to get a stream of StreamingInputCallRequest from the client and return a single StreamingInputCallResponse with extra information.
This method does not have a request parameter at all. Instead, it uses an ExampleService_StreamingInputCallServer stream to both read and write messages. It uses the Recv() method to receive client messages, and returns a single response using the SendAndClose() method.
In this method body the ExampleService_StreamingInputCallServer Recv() method is used to repeatedly read in the client requests to a request object (in this case a StreamingInputCallResponse) until there are no more messages.
The server needs to check the error returned from the Recv() method after each call. If nil is returned the stream is still active and it is OK to continue reading. if io.EOF is returned, the message stream has ended and the server can return its response. If it has any other value, the error is returned “as is” so that it will be translated to an RPC status by the gRPC layer.
/* RPC that represent a sequence of requests and a single responses
* The server returns the aggregated size of client payload as the result.
*/
func (exampleServer) StreamingInputCall (stream example.ExampleService_StreamingInputCallServer) error {
log.Log_info("\n\nExampleService_StreamingInputCallServer:")
log.Log_dump_structure(stream)
/* Starting/ Closing a Client Stream. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"StreamingInputCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"StreamingInputCall")
size := 0
for {
req, err := stream.Recv()
if err == io.EOF {
return stream.SendAndClose(&example.StreamingInputCallResponse{
AggregatedPayloadSize: int32(size),
})
}
size += len(req.User.Name)
if err != nil {
return err
}
}
}
Bidirectional Streaming RPC
The bidirectional streaming RPC FullDuplexCall() allows one request to produce multiple responses, using full duplexing.
/* RPC that represent a sequence of requests and responses
* with each request served by the server immediately.
* As one request could lead to multiple responses, this interface
* demonstrates the idea of full duplexing.
*/
func (exampleServer) FullDuplexCall (stream example.ExampleService_FullDuplexCallServer) error {
log.Log_info("\n\nExampleService_FullDuplexCallServer:")
log.Log_dump_structure(stream)
/* Starting/ Closing a Client/ Server Streams. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"FullDuplexCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"FullDuplexCall")
for {
req, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return status.Error(codes.Internal, err.Error())
}
if req.ResponseStatus != nil && req.ResponseStatus.Code != int32(codes.OK) {
return status.Error(codes.Code(req.ResponseStatus.Code), "error")
}
resp := &example.StreamingOutputCallResponse{User: &example.User{}}
for _, param := range req.ResponseParameters {
if stream.Context().Err() != nil {
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += req.GetUser().GetName()
}
resp.User.Name = buf
if err := stream.Send(resp); err != nil {
return err
}
}
}
}
The ExampleService_FullDuplexCallServer stream is used to read and write messages. However, return values can be written while the client is still writing messages to the message stream.
The syntax for reading and writing here is very similar to the client-streaming method, except the server uses the Send() method rather than the SendAndClose() method. Each side will always get the other’s messages in the order they were written, and both the client and server can read and write completely independently.
The following example is also bidirectional streaming and uses the RPC HalfDuplexCall(). However, now the server buffers all the client requests and then serves them in order. A stream of responses are returned to the client when the server starts with first request. This interface demonstrates the idea of half duplexing.
/* RPC that represent a sequence of requests and responses.
* The server buffers all the client requests and then serves them in order.
* A stream of responses are returned to the client when the server starts with
* first request.
*/
func (exampleServer) HalfDuplexCall (stream example.ExampleService_HalfDuplexCallServer) error {
log.Log_info("\n\nExampleService_HalfDuplexCallServer:")
log.Log_debug_dump(stream)
/* Starting/ Closing a Client/ Server Streams. Update monitoring information. */
netconfd_connect.Open_streams("example",
"ExampleService",
"HalfDuplexCall")
defer netconfd_connect.Close_streams("example",
"ExampleService",
"HalfDuplexCall")
requests := []*example.StreamingOutputCallRequest{}
for {
req, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return status.Error(codes.Internal, err.Error())
}
requests = append(requests, req)
}
for _, req := range requests {
resp := &example.StreamingOutputCallResponse{User: &example.User{}}
for _, param := range req.ResponseParameters {
if stream.Context().Err() != nil {
return stream.Context().Err()
}
buf := ""
for i := 0; i < int(param.GetSize()); i++ {
buf += req.GetUser().GetName()
}
resp.User.Name = buf
if err := stream.Send(resp); err != nil {
return err
}
}
}
return nil
}
Starting gRPC Server
Once all the methods are implemented, a gRPC server needs to be started.
The following snippet shows how to start the RouteGuide service in the ypgrpc-go-app application:
/* All initialization is done
* Start the gRPC server to listen for clients
*/
lis, err := net.Listen("tcp", addr)
if err != nil {
log.Log_error("\nError: failed to Listen (%v)", err)
}
/* Start the gRPC server and Register all the gRPC Services */
grpcServer := grpc.NewServer()
/* Register All the Services here */
helloworld.RegisterGreeterServer(grpcServer, &helloworldServer{})
example.RegisterExampleServiceServer(grpcServer, &exampleServer{})
log.Log_info("\nypgrpc_server: Starting to serve")
if err := grpcServer.Serve(lis); err != nil {
log.Log_error("\nError: failed to Serve (%v)", err)
}
gRPC server implementation consist of the following steps:
Specify the port to listen for client requests with the --port CLI parameter (default is
50830)Create an instance of the gRPC server using grpc.NewServer(...)
Register the service implementation with the gRPC server.
Call Serve() on the server to set the port details