DEV Community: Toran Sahu

Call for PyCon India 2020

Toran Sahu — Thu, 25 Feb 2021 13:45:24 +0000

Hello All,

This is to brief you about PyCon.

PyCon is the largest premier Python conference hosted all around the world, mostly organized by volunteers from local Python communities.

So thus, PyCon India is the largest gathering of Pythonistas in India for the Python programming language. Which is scheduled to be held on Oct 2nd to Oct 5th on an online platform.

This year, with the unique scenario at hand, the conference has become more accessible to make your participation right from the comfort of your homes.

Schedule & Venue

When: Oct 2nd to Oct 5th, 2020
Where: Online
Website: in.pycon.org/2020

Participation

Whether you are an experienced programmer, a hobby hacker, or an absolute beginner, PyCon welcomes you to the Python community.

The conference happens in a very precise format each year. 2 days of talks on over different topics, then 2 days of hands-on which starts with a workshop followed by dev sprint.

Get your passes from here.

Volunteering

Have you ever wished you could contribute or give back to the Python community? Well, here’s an awesome opportunity for you to do it with the year’s premier conference for the Python programming language, PyCon India 2020.

You can join any of the following groups as per your interests and skill sets and contribute.

Media team
Design team
Content team
Technology team

You can find various open tasks on PyCon India Github account. A few topics which welcomes your contributions are:

Main Website: pythonindia/inpycon2020
Blog Page: pythonindia/inpycon-blog
Conference Management Tool: pythonindia/junction

Proposals

The conference welcomes talks of every level. Whether you are a beginner filling the proposal for your first talk or have been giving talks for the past 10+ years. It encourages everyone to draft a proposal, no matter the difficulty level.

You can learn more about it, get the current proposal list, and also submit your proposals here.

The best way to get the latest updates on the conference, connect with the PyCon India community, or to get answers to your queries faster, is joining the PyCon India Zulip Channel.

See you at the conference!

PyCon India 2019

Toran Sahu — Thu, 25 Feb 2021 13:31:29 +0000

Developer conferences excite many techies and I'm glad I fall into that category as well. — it gives an opportunity to talk to other techies around the country and even around the globe one-to-one, showcase and exchange the latest stuff that excites each other.

Associating with PyCon India was something I had been looking forward to, and why not? It is the largest premier Python conference in India on using and developing the Python programming language, attended by students, professionals and Python veterans all alike.

The conference took place in Chennai at Chennai Trade Center on October 12-15, 2019. About a 1600+ of the industry's Pythonistas from different fields were present there for the 11th edition of this mega event.

I arrived in the bustling city of Chennai, armed with my laptop, to soak up knowledge, listen to and interact with Pythonistas over there, as much as I could.

The Schedule

October 12-13

Opening Keynote
Total 19 Talks in 6 + 1 slots (3 talks in 3 halls simultaneously)
Lightning Talks
Closing Keynote

October 14-15

Workshops
Dev Sprints

The conference was of total 4 days - 2 days speech and 2 days hands-on. An usual speech day was like: We started at 8 O'Clock in the morning with registration and breakfast till 9:15 am. Gathered for a 15 minutes of Opening Address, and then attended an hour of Opening Keynote. Post than we had a small tea break to keep sleepies awaken :P. Then 3 slots of Talks were delivered by Pythonistas from various fields, companies, schools & colleges around the India. The each talks were half an hour long. This time it was organized in 3 different halls simultaneously to cover and deliver more and more of it. In these talks people shared their experience, knowledge, and their creation related to Python.

Then an additional talk for the interested ones. Then lunch of ~1 hour was followed by Lightning Talks, where people showcased their creations using working models demostration, and poster designs in an open area.

Post that, we had again 3 slots of Talks on different topics, followed by a tea break. And before the Closing Address for the day, we got most awaited Closing Keynote of the day by one of the famous guest speakers.

I couldn't attend Workshops and Dev Sprints this time but will definitely plan for the next one.

Day 1

The first of the conference, and opening keynote of the day was Python’s journey to Data Science by Jake VanderPlas, author of Python Data Science Handbook, an Astronomer, and many more.

The keynote started with the beginning of Python and insights on roadmaps of the language so far. Jake walked us through the various era of Python programming language - How Python became a data science powerhouse - Various data science tools available in Python till date and so on.

frame sharing with Jake VanderPlas

The closing keynote emphasized on We Shall Overcome - Gender gap in tech by Devi A. S. L., Staff Engineer at RazorPay.

"Looking around me, it is so obvious that women dont't get equal opportunities in tech and there are so many things that lead to that." - Guido van Rossum, 2014

Devi is a very senior Pythonista and she has been active in the community since the inception of PyCon India. She started her speech with a small story of her which shed some light on Women Empowerment in tech. She strongly put her points with the facts about what is happening wrong since long time and how we can overcome that.

Day 2

The opening keynote of the day was Let Them Write Code by Ines Montani, founder of Explosion, core contributor of spaCy, prodigy.

The keynote started with a brief introduction about Ines, her company, and about tools - spaCy and prodigy. She also demonstrated how simple and helpful prodigy is for tokenization purposes in data science field.

And the closing keynote was A talk near the future of Python by David Beazley, author of Python Cookbook.

This was the most awaited keynote. Audiences were eagerly waiting to listen to him after 3 slots of small talks. The hall was full packed and was echoing loud due to tremendous applauses and cheerings. He started reading a note he has wrote in late 1984 (February 14th :<3) about computers, machine language, and a game he was writing for Apple II in machine language. Very interesting.. Very exciting!

While talking these stuffs and about his love for machines.. he opened his Emac and started writing a Python class to implement a stack.. and rest was history. It turned out that he live-coded an interpreter for WebAssembly in Python and ran a rocket game's (The game written by someone else in Rust using WebAssembly) binary code in a browser using that interpreter.

It took him barely an hour to write a viable WebAssembly interpreter in Python. Must watch his live-code session.

frame sharing with David Beazley

Why should you attend PyCon

To get exposure to the actionable best practices & tools that you could use in our projects, saving money and time spent exploring different options.
To get a sneak peek at upcoming innovations.
PyCon, provides a platform to unmask the rationality behind the trending technologies like ML/AI, Automation, Web development, Software development, IOT and more that use Python.
To learn technologies from a wide array of talks, workshops, keynotes speech, developer sprints each presented by inspiring and enthusiastic industry experts, community leaders and many young minds delving into new technologies.
To get an opportunity to network with like-minded people from diverse backgrounds including geek programmers, enthusiasts, entrepreneurs, scholars, intellectuals and many more.
A chance to spread the word about your company and create goodwill about the opportunities and incentives the employees are given. This will be the ideal way to attract future employees from an diversified talent pool.

A few days later

I got a certificate of participation from PyCon India.

I had a great time at PyCon India 2019. I came back with a lot of new ideas, knowldege, and experience which will soon make it to my current and upcoming projects. Keep following me to meet me at the next conference near you!

Part-3: Building a basic microservice with bidirectional-streaming gRPC using Golang

Toran Sahu — Thu, 25 Feb 2021 11:16:13 +0000

If you have been through part-2 of this blog series, you would have already got to know that the gRPC framework has got support for uni-directional streaming RPCs. But that is not the end. gRPC has support for bi-directional RPCs as well. Being said that, a gRPC client and a gRPC server can stream requests and responses simultaneously utilizing the same TCP connection.

Objective

In this blog, you'll get to know what is bi-directional streaming RPCs. How to implement, test, and run them using a live, fully functional example.

Previously in the part-2 of this blog series, we've learned the basics of uni-directional streaming gRPCs, how to implement those gRPC, how to write unit tests, how to launch the server & client. part-2 walks you through a step-by-step guide to implement a Stack Machine server & client leveraging the uni-directional streaming RPC.

If you've missed that, it is highly recommended to go through it to get familiar with the basics of the gRPC framework & streaming RPCs.

Introduction

Let's understand how bi-directional streaming RPCs works at a very high-level.

Bidirectional streaming RPCs 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 best thing is, the order of messages in each stream is preserved.

Now let's improve the "Stack Machine" server & client codes to support bidirectional streaming.

Implementing Bidirectional Streaming RPC

We already have Server-streaming RPC ServerStreamingExecute() to handle FIB operation which streams the numbers from Fibonacci series, and Client-streaming RPC Execute() to handle the stream of instructions to perform basic ADD/SUB/MUL/DIV operations and return a single response.

In this blog we'll merge both the functionality to make the Execute() RPC a bidirectional streaming one.

Update the protobuf

Let's update the machine/machine.proto to make Execute() a bi-directional (server & client streaming) RPC, so that the client can stream the instructions rather than sending a set of instructions to the server & the server can respond with a stream of results. Doing so, we're getting rid of InstructionSet and ServerStreamingExecute().

The updated machine/machine.proto now looks like:

 service Machine {
-     rpc Execute(stream Instruction) returns (Result) {}
-     rpc ServerStreamingExecute(InstructionSet) returns (stream Result) {}
+     rpc Execute(stream Instruction) returns (stream Result) {}
 }

source: machine/machine.proto

Notice the stream keyword at two places in the Execute() RPC declaration.

Generating the updated client and server interface Go code

Now lets generate an updated golang code from the machine/machine.proto by running:

~/disk/E/workspace/grpc-eg-go
$ SRC_DIR=./
$ DST_DIR=$SRC_DIR
$ protoc \
  -I=$SRC_DIR \
  --go_out=plugins=grpc:$DST_DIR \
  $SRC_DIR/machine/machine.proto

You'll notice that declaration of ServerStreamingExecute() has been removed from MachineServer & MachineClient interfaces. However the signature of Execute() is intact.

...

 type MachineServer interface {
    Execute(Machine_ExecuteServer) error
-   ServerStreamingExecute(*InstructionSet, Machine_ServerStreamingExecuteServer) error
 }

...

 type MachineClient interface {
    Execute(ctx context.Context, opts ...grpc.CallOption) (Machine_ExecuteClient, error)
-   ServerStreamingExecute(ctx context.Context, in *InstructionSet, opts ...grpc.CallOption) (Machine_ServerStreamingExecuteClient, error)
 }

...

source: machine/machine.pb.go

Update the Server

Now we need to update the server code to make Execute() a bi-directional (server & client streaming) RPC so that it should be able to accept stream the instructions from the client and at the same time it can respond with a stream of results.

func (s *MachineServer) Execute(stream machine.Machine_ExecuteServer) error {
    var stack stack.Stack
    for {
        instruction, err := stream.Recv()
        if err == io.EOF {
            log.Println("EOF")
            return nil
        }
        if err != nil {
            return err
        }

        operand := instruction.GetOperand()
        operator := instruction.GetOperator()
        op_type := OperatorType(operator)

        fmt.Printf("Operand: %v, Operator: %v\n", operand, operator)

        switch op_type {
        case PUSH:
            stack.Push(float32(operand))
        case POP:
            stack.Pop()
        case ADD, SUB, MUL, DIV:
            item2, popped := stack.Pop()
            item1, popped := stack.Pop()

            if !popped {
                return status.Error(codes.Aborted, "Invalid sets of instructions. Execution aborted")
            }

            var res float32
            if op_type == ADD {
                res = item1 + item2
            } else if op_type == SUB {
                res = item1 - item2
            } else if op_type == MUL {
                res = item1 * item2
            } else if op_type == DIV {
                res = item1 / item2
            }

            stack.Push(res)
            if err := stream.Send(&machine.Result{Output: float32(res)}); err != nil {
                return err
            }
        case FIB:
            n, popped := stack.Pop()

            if !popped {
                return status.Error(codes.Aborted, "Invalid sets of instructions. Execution aborted")
            }

            if op_type == FIB {
                for f := range utils.FibonacciRange(int(n)) {
                    if err := stream.Send(&machine.Result{Output: float32(f)}); err != nil {
                        return err
                    }
                }
            }
        default:
            return status.Errorf(codes.Unimplemented, "Operation '%s' not implemented yet", operator)
        }
    }
}

source: server/machine.go

Update the Client

Let's update the client code to make client.Execute() a bi-directional streaming RPC so that the client can stream the instructions to the server and can receive a stream of results at the same time.

func runExecute(client machine.MachineClient, instructions []*machine.Instruction) {
    log.Printf("Streaming %v", instructions)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    stream, err := client.Execute(ctx)
    if err != nil {
        log.Fatalf("%v.Execute(ctx) = %v, %v: ", client, stream, err)
    }
    waitc := make(chan struct{})
    go func() {
        for {
            result, err := stream.Recv()
            if err == io.EOF {
                log.Println("EOF")
                close(waitc)
                return
            }
            if err != nil {
                log.Printf("Err: %v", err)
            }
            log.Printf("output: %v", result.GetOutput())
        }
    }()

    for _, instruction := range instructions {
        if err := stream.Send(instruction); err != nil {
            log.Fatalf("%v.Send(%v) = %v: ", stream, instruction, err)
        }
        time.Sleep(500 * time.Millisecond)
    }
    if err := stream.CloseSend(); err != nil {
        log.Fatalf("%v.CloseSend() got error %v, want %v", stream, err, nil)
    }
    <-waitc
}

source: client/machine.go

Test

Before we start updating the unit test, let's generate mocks for MachineClient, Machine_ExecuteClient, and Machine_ExecuteServer interfaces to mock the stream type while testing the bidirectional streaming RPC Execute().

~/disk/E/workspace/grpc-eg-go
$ mockgen github.com/toransahu/grpc-eg-go/machine MachineClient,Machine_ExecuteServer,Machine_ExecuteClient > mock_machine/machine_mock.go

The updated mock_machine/machine_mock.go should look like this.

Now, we're good to write a unit test for bidirectional streaming RPC Execute().

Server

Let's update the unit test to test the server-side logic of Execute() RPC for bidirectional streaming using mock:

func TestExecute(t *testing.T) {
    s := MachineServer{}

    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockServerStream := mock_machine.NewMockMachine_ExecuteServer(ctrl)

    mockResults := []*machine.Result{}
    callRecv1 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operand: 1, Operator: "PUSH"}, nil)
    callRecv2 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operand: 2, Operator: "PUSH"}, nil).After(callRecv1)
    callRecv3 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operator: "MUL"}, nil).After(callRecv2)
    callRecv4 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operand: 3, Operator: "PUSH"}, nil).After(callRecv3)
    callRecv5 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operator: "ADD"}, nil).After(callRecv4)
    callRecv6 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operator: "FIB"}, nil).After(callRecv5)
    mockServerStream.EXPECT().Recv().Return(nil, io.EOF).After(callRecv6)
    mockServerStream.EXPECT().Send(gomock.Any()).DoAndReturn(
        func(result *machine.Result) error {
            mockResults = append(mockResults, result)
            return nil
        }).AnyTimes()
    wants := []float32{2, 5, 0, 1, 1, 2, 3, 5}

    err := s.Execute(mockServerStream)
    if err != nil {
        t.Errorf("Execute(%v) got unexpected error: %v", mockServerStream, err)
    }
    for i, result := range mockResults {
        got := result.GetOutput()
        want := wants[i]
        if got != want {
            t.Errorf("got %v, want %v", got, want)
        }
    }
}

source: server/machine_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test server/machine.go server/machine_test.go
ok      command-line-arguments  0.004s

Client

Now, add unit test to test client-side logic of Execute() RPC for bidirectional streaming using mock:

func testExecute(t *testing.T, client machine.MachineClient) {
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()

    instructions := []*machine.Instruction{}
    instructions = append(instructions, &machine.Instruction{Operand: 5, Operator: "PUSH"})
    instructions = append(instructions, &machine.Instruction{Operand: 6, Operator: "PUSH"})
    instructions = append(instructions, &machine.Instruction{Operator: "MUL"})

    stream, err := client.Execute(ctx)
    if err != nil {
        log.Fatalf("%v.Execute(%v) = _, %v: ", client, ctx, err)
    }
    for _, instruction := range instructions {
        if err := stream.Send(instruction); err != nil {
            log.Fatalf("%v.Send(%v) = %v: ", stream, instruction, err)
        }
    }
    result, err := stream.Recv()
    if err != nil {
        log.Fatalf("%v.Recv() got error %v, want %v", stream, err, nil)
    }

    got := result.GetOutput()
    want := float32(30)
    if got != want {
        t.Errorf("got %v, want %v", got, want)
    }
}

func TestExecute(t *testing.T) {
    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockMachineClient := mock_machine.NewMockMachineClient(ctrl)

    mockClientStream := mock_machine.NewMockMachine_ExecuteClient(ctrl)
    mockClientStream.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
    mockClientStream.EXPECT().Recv().Return(&machine.Result{Output: 30}, nil)

    mockMachineClient.EXPECT().Execute(
        gomock.Any(), // context
    ).Return(mockClientStream, nil)

    testExecute(t, mockMachineClient)
}

source: mock_machine/machine_mock_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test mock_machine/machine_mock.go mock_machine/machine_mock_test.go
ok      command-line-arguments  0.003s

Run

Now we are assured through unit tests that the business logic of the server & client codes is working as expected, let’s try running the server and communicating to it via our client code.

Server

To spin up the server we need to run the previously created cmd/run_machine_server.go file.

~/disk/E/workspace/grpc-eg-go
$ go run cmd/run_machine_server.go

Client

Now, let’s run the client code client/machine.go.

~/disk/E/workspace/grpc-eg-go
$ go run client/machine.go
Streaming [operator:"PUSH" operand:1  operator:"PUSH" operand:2  operator:"ADD"  operator:"PUSH" operand:3  operator:"DIV"  operator:"PUSH" operand:4  operator:"MUL"  operator:"FIB"  operator:"PUSH" operand:5  operator:"PUSH" operand:6  operator:"SUB" ]
output: 3
output: 1
output: 4
output: 0
output: 1
output: 1
output: 2
output: 3
output: -1
EOF

Bonus

There are situations when one has to choose between mocking a dependency versus incorporating the dependencies into the test environment & running them live.

The decision - whether to mock or not could be made based on:

how many dependencies are there
which are the essential & most used dependencies
is it feasible to install dependencies on the test (and even the developer's) environment, etc.

To one extreme we can mock everything. But the mocking effort should pay us off.

For the gRPC framework, we can run the gRPC server live & write client codes to test against the business logic.
But spinning up a server from the test file can lead to unintended consequences that may require you to allocate a TCP port (parallel runs, multiple runs under the same CI server).

To solve this gRPC community has introduced a package called bufconn under gRPC's testing package. bufconn is a package that provides a Listener object that implements net.Conn. We can substitute this listener in a gRPC server - allowing us to spin up a server that acts as a full-fledged server that can be used for testing that talks over an in-memory buffer instead of a real port.

As bufconn already comes with the grpc go module - which we already have installed, we don't need to install it explicitly.

So, let's create a new test file server/machine_live_test.go write the following test code to launch the gRPC server live using bufconn, and write a client to test the bidirectional RPC Execute().

const bufSize = 1024 * 1024

var lis *bufconn.Listener

func init() {
    lis = bufconn.Listen(bufSize)
    s := grpc.NewServer()
    machine.RegisterMachineServer(s, &MachineServer{})
    go func() {
        if err := s.Serve(lis); err != nil {
            log.Fatalf("Server exited with error: %v", err)
        }
    }()
}

func bufDialer(context.Context, string) (net.Conn, error) {
    return lis.Dial()
}

func testExecute_Live(t *testing.T, client machine.MachineClient, instructions []*machine.Instruction, wants []float32) {
    log.Printf("Streaming %v", instructions)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    stream, err := client.Execute(ctx)
    if err != nil {
        log.Fatalf("%v.Execute(ctx) = %v, %v: ", client, stream, err)
    }
    waitc := make(chan struct{})
    go func() {
        i := 0
        for {
            result, err := stream.Recv()
            if err == io.EOF {
                log.Println("EOF")
                close(waitc)
                return
            }
            if err != nil {
                log.Printf("Err: %v", err)
            }
            log.Printf("output: %v", result.GetOutput())
            got := result.GetOutput()
            want := wants[i]
            if got != want {
                t.Errorf("got %v, want %v", got, want)
            }
            i++
        }
    }()

    for _, instruction := range instructions {
        if err := stream.Send(instruction); err != nil {
            log.Fatalf("%v.Send(%v) = %v: ", stream, instruction, err)
        }
    }
    if err := stream.CloseSend(); err != nil {
        log.Fatalf("%v.CloseSend() got error %v, want %v", stream, err, nil)
    }
    <-waitc
}

func TestExecute_Live(t *testing.T) {
    ctx := context.Background()
    conn, err := grpc.DialContext(ctx, "bufnet", grpc.WithContextDialer(bufDialer), grpc.WithInsecure())
    if err != nil {
        t.Fatalf("Failed to dial bufnet: %v", err)
    }
    defer conn.Close()
    client := machine.NewMachineClient(conn)

    // try Execute()
    instructions := []*machine.Instruction{
        {Operand: 1, Operator: "PUSH"},
        {Operand: 2, Operator: "PUSH"},
        {Operator: "ADD"},
        {Operand: 3, Operator: "PUSH"},
        {Operator: "DIV"},
        {Operand: 4, Operator: "PUSH"},
        {Operator: "MUL"},
        {Operator: "FIB"},
        {Operand: 5, Operator: "PUSH"},
        {Operand: 6, Operator: "PUSH"},
        {Operator: "SUB"},
    }
    wants := []float32{3, 1, 4, 0, 1, 1, 2, 3, -1}
    testExecute_Live(t, client, instructions, wants)
}

source: server/machine_live_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test server/machine.go server/machine_live_test.go
ok      command-line-arguments  0.005s

Fantastic!!! Everything worked as expected.

At the end of this blog, we’ve learned:

How to define an interface for bi-directional streaming RPC using protobuf
How to write gRPC server & client logic for bi-directional streaming RPC
How to write and run the unit test for bi-directional streaming RPC
How to write and run the unit test for bi-directional streaming RPC by running the server live leveraging the bufconn package
How to run the gRPC server and a client can communicate to it

The source code of this example is available at toransahu/grpc-eg-go.

See you on the next blog.

Part-2: Building a basic microservice with unidirectional-streaming gRPC using Golang

Toran Sahu — Thu, 25 Feb 2021 11:14:24 +0000

Have you ever wondered while developing a REST API that if the server could have got the capability to stream responses using the same TCP connection? Or, reversely if the REST client could have got the capability to stream the requests to the server, this could have saved the cost of bringing up another service (like WebSocket) just for the sake of fulfilling such requirement.

Then REST isn’t the only API architecture available, and for such use-cases, the gRPC model has begun to play a crucial role. gRPC's unidirectional-streaming RPC feature has got your back on those requirements.

Objective

In this blog, you'll get to know what is client streaming & server streaming uni-directional RPCs. How to implement, test, and run them using a live, fully functional example.

Previously in the part-1 of this blog series, we've learned the basics of gRPC, how to implement a Simple/Unary gRPC, how to write unit tests, how to launch the server & client. part-1 walks you through a step-by-step guide to implement a Stack Machine server & client leveraging Simple/Unary RPC.

If you've missed that, it is highly recommended to go through it to get familiar with the basics of the gRPC framework.

Introduction

Let's understand how Client streaming & Server streaming RPCs works at a very high-level.

Client streaming RPCs where:

the client writes a sequence of messages and sends them to the server using a provided stream
once the client has finished writing the messages, it waits for the server to read them and return its response

Server streaming RPCs 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

The best thing is gRPC guarantees message ordering within an individual RPC call.

Now let's improve the "Stack Machine" server & client codes to support unidirectional streaming.

Implementing Server Streaming RPC

We'll see an example of Server Streaming first by implementing the FIB operation.

Where the FIB RPC will:

perform a Fibonacci operation
accept an integer input i.e. generate first N numbers of the Fibonacci series
will respond with a stream of integers i.e. first N numbers of the Fibonacci series

And later we'll see how Client Streaming can be implemented so that a client can input a stream of Instructions to the Stack Machine in real-time rather than sending a single request comprised of a set of Instructions.

Update the protobuf

We already have defined the gRPC service Machine and a Simple (Unary) RPC method Execute inside our service definition in part-1 of the blog series. Now, let's update the service definition to add one server streaming RPC called ServerStreamingExecute.

A server streaming RPC where the client sends a request to the server using the stub and waits for a response to come back as a stream of result
To specify a server-side streaming method, need to place the stream keyword before the response type

// ServerStreamingExecute accepts a set of Instructions from client and returns a stream of Result.
rpc ServerStreamingExecute(InstructionSet) returns (stream Result) {}

source: machine/machine.proto

Generating the updated client and server interface Go code

We need to generate the gRPC client and server interfaces from our machine/machine.proto service definition.

~/disk/E/workspace/grpc-eg-go
$ SRC_DIR=./
$ DST_DIR=$SRC_DIR
$ protoc \
  -I=$SRC_DIR \
  --go_out=plugins=grpc:$DST_DIR \
  $SRC_DIR/machine/machine.proto

You can observe that the declaration of ServerStreamingExecute() in the MachineClient and MachineServer interface has been auto-generated:

...

 type MachineClient interface {
    Execute(ctx context.Context, in *InstructionSet, opts ...grpc.CallOption) (*Result, error)
+   ServerStreamingExecute(ctx context.Context, in *InstructionSet, opts ...grpc.CallOption) (Machine_ServerStreamingExecuteClient, error)
 }

...

 type MachineServer interface {
    Execute(context.Context, *InstructionSet) (*Result, error)
+   ServerStreamingExecute(*InstructionSet, Machine_ServerStreamingExecuteServer) error
 }

source: machine/machine.pb.go

Update the Server

Just in case if you're wondering, What if my service doesn't implement some of the RPCs declared in the machine.pb.go file, then you'll encounter the following error while launching your gRPC server.

~/disk/E/workspace/grpc-eg-go
$ go run cmd/run_machine_server.go
# command-line-arguments
cmd/run_machine_server.go:32:44: cannot use &server.MachineServer literal (type *server.MachineServer) as type machine.MachineServer in argument to machine.RegisterMachineServer:
        *server.MachineServer does not implement machine.MachineServer (missing ServerStreamingExecute method)

So, it's always the best practice to keep your service in sync with the service definition i.e. machine/machine.proto & machine/machine.pb.go. If you do not want to support a particular RPC, or its implementation is not yet ready, just respond with Unimplemented error status. Example:

// ServerStreamingExecute runs the set of instructions given and streams a sequence of Results.
func (s *MachineServer) ServerStreamingExecute(instructions *machine.InstructionSet, stream machine.Machine_ServerStreamingExecuteServer) error {
    return status.Error(codes.Unimplemented, "ServerStreamingExecute() not implemented yet")
}

source: server/machine.go

Before we implement the ServerStreamingExecute() RPC, let's write a Fibonacci series generator called FibonacciRange().

package utils

func FibonacciRange(n int) <-chan int {
    ch := make(chan int)
    fn := make([]int, n+1, n+2)
    fn[0] = 0
    fn[1] = 1
    go func() {
        defer close(ch)
        for i := 0; i <= n; i++ {
            var f int
            if i < 2 {
                f = fn[i]
            } else {
                f = fn[i-1] + fn[i-2]
            }
            fn[i] = f
            ch <- f
        }
    }()
    return ch
}

source: utils/fibonacci.go

The blog series assumes that you're familiar with Golang basics & its concurrency paradigms & concepts like Channels. You can read more about the Channels from the official document.

This function yields the numbers of Fibonacci series till the Nth position.

Let's also add a small unit test to validate the FibonacciRange() generator.

package utils

import (
    "testing"
)

func TestFibonacciRange(t *testing.T) {
    fibOf5 := []int{0, 1, 1, 2, 3, 5}
    i := 0
    for f := range FibonacciRange(5) {
        if f != fibOf5[i] {
            t.Errorf("got %d, want %d", f, fibOf5[i])
        }
        i++
    }
}

source: utils/fibonacci_test.go

Let's implement ServerStreamingExecute() to handle the basic instructions PUSH/POP, and FIB with proper error handling. On completion of the execution of instructions set, it should POP the result from the Stack and should respond with a Result object to the client.

func (s *MachineServer) ServerStreamingExecute(instructions *machine.InstructionSet, stream machine.Machine_ServerStreamingExecuteServer) error {
    if len(instructions.GetInstructions()) == 0 {
        return status.Error(codes.InvalidArgument, "No valid instructions received")
    }

    var stack stack.Stack

    for _, instruction := range instructions.GetInstructions() {
        operand := instruction.GetOperand()
        operator := instruction.GetOperator()
        op_type := OperatorType(operator)

        log.Printf("Operand: %v, Operator: %v\n", operand, operator)

        switch op_type {
        case PUSH:
            stack.Push(float32(operand))
        case POP:
            stack.Pop()
        case FIB:
            n, popped := stack.Pop()

            if !popped {
                return status.Error(codes.Aborted, "Invalid sets of instructions. Execution aborted")
            }

            if op_type == FIB {
                for f := range utils.FibonacciRange(int(n)) {
                    log.Println(float32(f))
                    stream.Send(&machine.Result{Output: float32(f)})
                }
            }
        default:
            return status.Errorf(codes.Unimplemented, "Operation '%s' not implemented yet", operator)
        }
    }
    return nil
}

source: server/machine.go

Update the Client

Now, update the client code to call ServerStreamingExecute() where the client will be receiving numbers of the Fibonacci series through the stream and print the same.

func runServerStreamingExecute(client machine.MachineClient, instructions *machine.InstructionSet) {
    log.Printf("Executing %v", instructions)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    stream, err := client.ServerStreamingExecute(ctx, instructions)
    if err != nil {
        log.Fatalf("%v.Execute(_) = _, %v: ", client, err)
    }
    for {
        result, err := stream.Recv()
        if err == io.EOF {
            log.Println("EOF")
            break
        }
        if err != nil {
            log.Printf("Err: %v", err)
            break
        }
        log.Printf("output: %v", result.GetOutput())
    }
    log.Println("DONE!")
}

source: client/machine.go

Test

To write the unit test we'll need to generate the mock of multiple interface as required.
mockgen is the ready-to-go framework for mocking in Golang, so we'll be leveraging it in our unit tests.

Server

As we've upgdated our interface i.e. machine/machine.pb.go, let's update the mock for MachineClient interface. And as we've introduced a new RPC ServerStreamingExecute(), let's generate the mock for ServerStream interface Machine_ServerStreamingExecuteServer as well.

~/disk/E/workspace/grpc-eg-go
$ mockgen github.com/toransahu/grpc-eg-go/machine MachineClient,Machine_ServerStreamingExecuteServer > mock_machine/machine_mock.go

The updated mock_machine/machine_mock.go should look like this.

Now, we're good to write unit test for server-side streaming RPC ServerStreamingExecute():

func TestServerStreamingExecute(t *testing.T) {
    s := MachineServer{}

    // set up test table
    tests := []struct {
        instructions []*machine.Instruction
        want         []float32
    }{
        {
            instructions: []*machine.Instruction{
                {Operand: 5, Operator: "PUSH"},
                {Operator: "FIB"},
            },
            want: []float32{0, 1, 1, 2, 3, 5},
        },
        {
            instructions: []*machine.Instruction{
                {Operand: 6, Operator: "PUSH"},
                {Operator: "FIB"},
            },
            want: []float32{0, 1, 1, 2, 3, 5, 8},
        },
    }

    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockServerStream := mock_machine.NewMockMachine_ServerStreamingExecuteServer(ctrl)
    for _, tt := range tests {
        mockResults := []*machine.Result{}
        mockServerStream.EXPECT().Send(gomock.Any()).DoAndReturn(
            func(result *machine.Result) error {
                mockResults = append(mockResults, result)
                return nil
            }).AnyTimes()

        req := &machine.InstructionSet{Instructions: tt.instructions}

        err := s.ServerStreamingExecute(req, mockServerStream)
        if err != nil {
            t.Errorf("ServerStreamingExecute(%v) got unexpected error: %v", req, err)
        }
        for i, result := range mockResults {
            got := result.GetOutput()
            want := tt.want[i]
            if got != want {
                t.Errorf("got %v, want %v", got, want)
            }
        }
    }
}

source: server/machine_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test server/machine.go server/machine_test.go
ok      command-line-arguments  0.003s

Client

For our new RPC ServerStreamingExecute(), let's add the mock for ClientStream interface Machine_ServerStreamingExecuteClient as well.

~/disk/E/workspace/grpc-eg-go
$ mockgen github.com/toransahu/grpc-eg-go/machine MachineClient,Machine_ServerStreamingExecuteServer,Machine_ServerStreamingExecuteClient > mock_machine/machine_mock.go

source: mock_machine/machine_mock.go

Let's add unit test to test client-side logic for server-side streaming RPC ServerStreamingExecute() using mock MockMachine_ServerStreamingExecuteClient :

func TestServerStreamingExecute(t *testing.T) {
    instructions := []*machine.Instruction{}
    instructions = append(instructions, &machine.Instruction{Operand: 1, Operator: "PUSH"})
    instructions = append(instructions, &machine.Instruction{Operator: "FIB"})
    instructionSet := &machine.InstructionSet{Instructions: instructions}

    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockMachineClient := mock_machine.NewMockMachineClient(ctrl)
    clientStream := mock_machine.NewMockMachine_ServerStreamingExecuteClient(ctrl)

    clientStream.EXPECT().Recv().Return(&machine.Result{Output: 0}, nil)

    mockMachineClient.EXPECT().ServerStreamingExecute(
        gomock.Any(),   // context
        instructionSet, // rpc uniary message
    ).Return(clientStream, nil)

    if err := testServerStreamingExecute(t, mockMachineClient, instructionSet); err != nil {
        t.Fatalf("Test failed: %v", err)
    }
}

source: mock_machine/machine_mock_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test mock_machine/machine_mock.go mock_machine/machine_mock_test.go
ok      command-line-arguments  0.003s

Run

As we are assured through unit tests that the business logic of the server & client codes is working as expected, let’s try running the server and communicating to it via our client code.

Server

To start the server we need to run the previously created cmd/run_machine_server.go file.

~/disk/E/workspace/grpc-eg-go
$ go run cmd/run_machine_server.go

Client

Now, let’s run the client code client/machine.go.

~/disk/E/workspace/grpc-eg-go
$ go run client/machine.go
Executing instructions:<operator:"PUSH" operand:5 > instructions:<operator:"PUSH" operand:6 > instructions:<operator:"MUL" >
output:30
Executing instructions:<operator:"PUSH" operand:6 > instructions:<operator:"FIB" >
output: 0
output: 1
output: 1
output: 2
output: 3
output: 5
output: 8
EOF
DONE!

Awesome! A Server Streaming RPC has been successfully implemented.

Implementing Client Streaming RPC

We have learned how to implement a Server Streaming RPC, now it's time to explore the Client Streaming RPC.
To do so, we'll not introduce another RPC, rather we'll update the existing Execute() RPC to accept a stream of Instructions from the client in real-time rather than sending a single request comprised of a set of Instructions.

Update the protobuf

So, let's update the interface:

 service Machine {
-     rpc Execute(InstructionSet) returns (Result) {}
+     rpc Execute(stream Instruction) returns (Result) {}
      rpc ServerStreamingExecute(InstructionSet) returns (stream Result) {}
 }

source: machine/machine.proto

Generating the updated client and server interface Go code

Now lets generate an updated golang code from the machine/machine.proto by running:

~/disk/E/workspace/grpc-eg-go
$ SRC_DIR=./
$ DST_DIR=$SRC_DIR
$ protoc \
  -I=$SRC_DIR \
  --go_out=plugins=grpc:$DST_DIR \
  $SRC_DIR/machine/machine.proto

You'll notice that declaration of Execute() has been updated from MachineServer & MachineClient interfaces.

 type MachineServer interface {
-   Execute(context.Context, *InstructionSet) (*Result, error)
+   Execute(Machine_ExecuteServer) error
    ServerStreamingExecute(*InstructionSet, Machine_ServerStreamingExecuteServer) error
 }

 type MachineClient interface {
-    Execute(ctx context.Context, in *InstructionSet, opts ...grpc.CallOption) (*Result, error)
+    Execute(ctx context.Context, opts ...grpc.CallOption) (Machine_ExecuteClient, error)
     ServerStreamingExecute(ctx context.Context, in *InstructionSet, opts ...grpc.CallOption) (Machine_ServerStreamingExecuteClient, error)
 }

source: machine/machine.pb.go

Update the Server

Let's update the server code to make Execute() a client streaming uni-directional RPC so that it should be able to accept stream the instructions from the client and respond with a Result struct.

func (s *MachineServer) Execute(stream machine.Machine_ExecuteServer) error {
    var stack stack.Stack
    for {
        instruction, err := stream.Recv()
        if err == io.EOF {
            log.Println("EOF")
            output, popped := stack.Pop()
            if !popped {
                return status.Error(codes.Aborted, "Invalid sets of instructions. Execution aborted")
            }

            if err := stream.SendAndClose(&machine.Result{
                Output: output,
            }); err != nil {
                return err
            }

            return nil
        }
        if err != nil {
            return err
        }

        operand := instruction.GetOperand()
        operator := instruction.GetOperator()
        op_type := OperatorType(operator)

        fmt.Printf("Operand: %v, Operator: %v\n", operand, operator)

        switch op_type {
        case PUSH:
            stack.Push(float32(operand))
        case POP:
            stack.Pop()
        case ADD, SUB, MUL, DIV:
            item2, popped := stack.Pop()
            item1, popped := stack.Pop()

            if !popped {
                return status.Error(codes.Aborted, "Invalid sets of instructions. Execution aborted")
            }

            if op_type == ADD {
                stack.Push(item1 + item2)
            } else if op_type == SUB {
                stack.Push(item1 - item2)
            } else if op_type == MUL {
                stack.Push(item1 * item2)
            } else if op_type == DIV {
                stack.Push(item1 / item2)
            }

        default:
            return status.Errorf(codes.Unimplemented, "Operation '%s' not implemented yet", operator)
        }
    }
}

source: server/machine.go

Update the Client

Now update the client code to make client.Execute() a uni-directional streaming RPC, so that the client can stream the instructions to the server and can receive a Result struct once the streaming completes.

func runExecute(client machine.MachineClient, instructions *machine.InstructionSet) {
    log.Printf("Streaming %v", instructions)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    stream, err := client.Execute(ctx)
    if err != nil {
        log.Fatalf("%v.Execute(ctx) = %v, %v: ", client, stream, err)
    }
    for _, instruction := range instructions.GetInstructions() {
        if err := stream.Send(instruction); err != nil {
            log.Fatalf("%v.Send(%v) = %v: ", stream, instruction, err)
        }
    }
    result, err := stream.CloseAndRecv()
    if err != nil {
        log.Fatalf("%v.CloseAndRecv() got error %v, want %v", stream, err, nil)
    }
    log.Println(result)
}

source: client/machine.go

Test

Generate mock for Machine_ExecuteClient and Machine_ExecuteServer interface to test client-streaming RPC Execute():

~/disk/E/workspace/grpc-eg-go
$ mockgen github.com/toransahu/grpc-eg-go/machine MachineClient,Machine_ServerStreamingExecuteClient,Machine_ServerStreamingExecuteServer,Machine_ExecuteServer,Machine_ExecuteClient > mock_machine/machine_mock.go

The updated mock_machine/machine_mock.go should look like this.

Server

Let's update the unit test to test the server-side logic of client streaming Execute() RPC using mock:

func TestExecute(t *testing.T) {
    s := MachineServer{}

    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockServerStream := mock_machine.NewMockMachine_ExecuteServer(ctrl)

    mockResult := &machine.Result{}
    callRecv1 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operand: 5, Operator: "PUSH"}, nil)
    callRecv2 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operand: 6, Operator: "PUSH"}, nil).After(callRecv1)
    callRecv3 := mockServerStream.EXPECT().Recv().Return(&machine.Instruction{Operator: "MUL"}, nil).After(callRecv2)
    mockServerStream.EXPECT().Recv().Return(nil, io.EOF).After(callRecv3)
    mockServerStream.EXPECT().SendAndClose(gomock.Any()).DoAndReturn(
        func(result *machine.Result) error {
            mockResult = result
            return nil
        })

    err := s.Execute(mockServerStream)
    if err != nil {
        t.Errorf("Execute(%v) got unexpected error: %v", mockServerStream, err)
    }
    got := mockResult.GetOutput()
    want := float32(30)
    if got != want {
        t.Errorf("got %v, wanted %v", got, want)
    }
}

source: server/machine_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test server/machine.go server/machine_test.go
ok      command-line-arguments  0.003s

Client

Now, add unit test to test client-side logic of client streaming Execute() RPC using mock:

func TestExecute(t *testing.T) {
    ctrl := gomock.NewController(t)
    defer ctrl.Finish()
    mockMachineClient := mock_machine.NewMockMachineClient(ctrl)

    mockClientStream := mock_machine.NewMockMachine_ExecuteClient(ctrl)
    mockClientStream.EXPECT().Send(gomock.Any()).Return(nil).AnyTimes()
    mockClientStream.EXPECT().CloseAndRecv().Return(&machine.Result{Output: 30}, nil)

    mockMachineClient.EXPECT().Execute(
        gomock.Any(), // context
    ).Return(mockClientStream, nil)

    testExecute(t, mockMachineClient)
}

source: mock_machine/machine_mock_test.go

Let's run the unit test:

~/disk/E/workspace/grpc-eg-go
$ go test mock_machine/machine_mock.go mock_machine/machine_mock_test.go
ok      command-line-arguments  0.003s

Run all the unit tests at once:

~/disk/E/workspace/grpc-eg-go
$ go test ./...
?       github.com/toransahu/grpc-eg-go/client  [no test files]
?       github.com/toransahu/grpc-eg-go/cmd     [no test files]
?       github.com/toransahu/grpc-eg-go/machine [no test files]
ok      github.com/toransahu/grpc-eg-go/mock_machine    (cached)
ok      github.com/toransahu/grpc-eg-go/server  (cached)
ok      github.com/toransahu/grpc-eg-go/utils   (cached)
?       github.com/toransahu/grpc-eg-go/utils/stack     [no test files]

Run

Now we are assured through unit tests that the business logic of the server & client codes is working as expected, let’s try running the server and communicating to it via our client code.

Server

To launch the server we need to run the previously created cmd/run_machine_server.go file.

~/disk/E/workspace/grpc-eg-go
$ go run cmd/run_machine_server.go

Client

Now, let’s run the client code client/machine.go.

~/disk/E/workspace/grpc-eg-go
$ go run client/machine.go
Streaming instructions:<operator:"PUSH" operand:5 > instructions:<operator:"PUSH" operand:6 > instructions:<operator:"MUL" >
output:30
Executing instructions:<operator:"PUSH" operand:6 > instructions:<operator:"FIB" >
output: 0
output: 1
output: 1
output: 2
output: 3
output: 5
output: 8
EOF
DONE!

Awesome!! We have successfully transformed a Unary RPC into Server Streaming RPC.

At the end of this blog, we’ve learned:

How to define an interface for uni-directional streaming RPCs using protobuf
How to write gRPC server & client logic for uni-directional streaming RPCs
How to write and run the unit test for server-streaming & client-streaming RPCs
How to run the gRPC server and a client can communicate to it

The source code of this example is available at toransahu/grpc-eg-go.
You can also git checkout to this commit SHA for Part-2(a) and to this commit SHA for Part-2(b).

See you in the next part of this blog series.

Part-1: Building a basic microservice with gRPC using Golang

Toran Sahu — Thu, 25 Feb 2021 11:08:17 +0000

Introduction

Being easy to use, REST API is the current most popular framework among developers for web application development. REST has been used to expose the services to the outer world, and also for internal communication among internal microservices. However, ease and flexibility come with some pitfalls. REST requires very strict Human Agreement, and relies on Documentation. Also, it has been found not so very performant in the case of internal communication and real-time applications. In 2015, gRPC kicked in. gRPC initially developed at Google is now disrupting the industry. gRPC is a modern open-source high-performance RPC framework, which comes with a simple language-agnostic Interface Definition Language (IDL) system, leveraging Protocol Buffers.

Objective

The purpose of this blog is to get you started with gRPC in Go with a simple working example. The blog covers basic information like What, Why, When/Where, and How about the gRPC. We'll majorly focus on the How section, to implement the client and server, to write unittests to test the client and server code separately, and will run the code to establish a client-server communication.

What is gRPC?

gRPC - Remote Procedure Call

gRPC is a high performance, open-source universal RPC Framework
It enables the server and client applications to communicate transparently and build connected systems
gRPC is developed and open-sourced by Google (but no, the g doesn’t stand for Google)

Why use gRPC?

Better Design
- With gRPC, we can define our service once in a .proto file and implement clients and servers in any of gRPC’s supported languages
- Ability to auto-generate and publish SDKs as opposed to publishing the APIs for services
High Performance
- Advantages of working with protocol buffers, including efficient serialization, a simple IDL, and easy interface updating
- Advantages of improved features of HTTP/2
- Multiplexing: This forces service-client to utilize a single TCP connection to simultaneously handle multiple requests
- Binary Framing and Compression
Multi-way communication
- Simple/Unary RPC
- Server-side streaming RPC
- Client-side streaming RPC
- Bidirectional streaming RPC

Where to use gRPC?

The “where” is pretty easy: we can leverage gRPC almost anywhere we have two computers communicating over a network:

Microservices
Client-Server Applications
Integrations and APIs
Browser-based Web Applications

How to use gRPC?

Our example is a simple “Stack Machine” as a service that lets clients perform operations like, PUSH, ADD, SUB, MUL, DIV, FIBB, AP, GP.

In Part-1 we’ll focus on Simple RPC implementation, in Part-2 we’ll focus on Server-side & Client-side streaming RPC, and in Part-3 we’ll implement Bidirectional streaming RPC.

Let's get started with installing the pre-requisites of the development.

Prerequisites

Go

Version 1.6 or higher.
For installation instructions, see Go’s Getting Started guide.

gRPC

Use the following command to install gRPC.

~/disk/E/workspace/grpc-eg-go
$ go get -u google.golang.org/grpc

Protocol Buffers v3

Install the protoc compiler that is used to generate gRPC service code. (https://developers.google.com/protocol-buffers/)

~/disk/E/workspace/grpc-eg-go
$ go get -u github.com/golang/protobuf/proto

Update the environment variable PATH to include the path to the protoc binary file.
Install the protoc plugin for Go

~/disk/E/workspace/grpc-eg-go
$ go get -u github.com/golang/protobuf/protoc-gen-go

Setting Project Structure

~/disk/E/workspace/grpc-eg-go
$ go mod init github.com/toransahu/grpc-eg-go
$ mkdir machine
$ mkdir server
$ mkdir client

$ tree
.
├── client/
├── go.mod
├── machine/
└── server/

Defining the service

Our first step is to define the gRPC service and the method request and response types using protocol buffers.

To define a service, we specify a named service in our machine/machine.proto file:

service Machine {
   ...
}

Then we define a Simple RPC method inside our service definition, specifying their request and response types.

A simple RPC where the client sends a request to the server using the stub and waits for a response to come back

// Execute accepts a set of Instructions from the client and returns a Result.
rpc Execute(InstructionSet) returns (Result) {}

machine/machine.proto file also contains protocol buffer message type definitions for all the request and response types used in our service methods.

// Result represents the output of execution of the instruction(s).
message Result {
  float output = 1;
}

Our machine/machine.proto file should look like this considering Part-1 of this blog series.

Generating client and server code

We need to generate the gRPC client and server interfaces from the machine/machine.proto service definition.

~/disk/E/workspace/grpc-eg-go
$ SRC_DIR=./
$ DST_DIR=$SRC_DIR
$ protoc \
  -I=$SRC_DIR \
  --go_out=plugins=grpc:$DST_DIR \
  $SRC_DIR/machine/machine.proto

Running this command generates the machine.pb.go file in the machine directory under the repository:

~/disk/E/workspace/grpc-eg-go
$ tree machine/
.
├── machine/
│   ├── machine.pb.go
│   └── machine.proto

Server

Let's create the server.

There are two parts to making our Machine service do its job:

Create server/machine.go: Implementing the service interface generated from our service definition; writing the business logic of our service.
Running the Machine gRPC server: Run the server to listen for requests from clients and dispatch them to the right service implementation.

Have a look how our MachineServer interface should look like: grpc-eg-go/server/machine.go

type MachineServer struct{}

// Execute runs the set of instructions given.
func (s *MachineServer) Execute(ctx context.Context, instructions *machine.InstructionSet) (*machine.Result, error) {
    return nil, status.Error(codes.Unimplemented, "Execute() not implemented yet")
}

Implementing Simple RPC

MachineServer implements only Execute() service method as of now - as per Part-1 of this blog series.
Execute(), just gets a InstructionSet from the client and returns the value in a Result by executing every Instruction in the InstructionSet into our Stack Machine.

Before implementing Execute(), let's implement a basic Stack. It should look like this.

type Stack []float32

func (s *Stack) IsEmpty() bool {
    return len(*s) == 0
}

func (s *Stack) Push(input float32) {
    *s = append(*s, input)
}

func (s *Stack) Pop() (float32, bool) {
    if s.IsEmpty() {
        return -1.0, false
    }
    item := (*s)[len(*s)-1]
    *s = (*s)[:len(*s)-1]
    return item, true
}

Now, let’s implement the Execute(). It should look like this.

type OperatorType string

const (
    PUSH OperatorType   = "PUSH"
    POP               = "POP"
    ADD               = "ADD"
    SUB               = "SUB"
    MUL               = "MUL"
    DIV               = "DIV"
)

type MachineServer struct{}

// Execute runs the set of instructions given.
func (s *MachineServer) Execute(ctx context.Context, instructions *machine.InstructionSet) (*machine.Result, error) {
    if len(instructions.GetInstructions()) == 0 {
        return nil, status.Error(codes.InvalidArgument, "No valid instructions received")
    }

    var stack stack.Stack

    for _, instruction := range instructions.GetInstructions() {
        operand := instruction.GetOperand()
        operator := instruction.GetOperator()
        op_type := OperatorType(operator)

        fmt.Printf("Operand: %v, Operator: %v", operand, operator)

        switch op_type {
        case PUSH:
            stack.Push(float32(operand))
        case POP:
            stack.Pop()
        case ADD, SUB, MUL, DIV:
            item2, popped := stack.Pop()
            item1, popped := stack.Pop()

            if !popped {
                return &machine.Result{}, status.Error(codes.Aborted, "Invalide sets of instructions. Execution aborted")
            }

            if op_type == ADD {
                stack.Push(item1 + item2)
            } else if op_type == SUB {
                stack.Push(item1 - item2)
            } else if op_type == MUL {
                stack.Push(item1 * item2)
            } else if op_type == DIV {
                stack.Push(item1 / item2)
            }

        default:
            return nil, status.Errorf(codes.Unimplemented, "Operation '%s' not implemented yet", operator)
        }

    }

    item, popped := stack.Pop()
    if !popped {
        return &machine.Result{}, status.Error(codes.Aborted, "Invalide sets of instructions. Execution aborted")
    }
    return &machine.Result{Output: item}, nil
}

We have implemented the Execute() to handle basic instructions like PUSH, POP, ADD, SUB, MUL, and DIV with proper error handling. On completion of the execution of the instructions set, it pops the result from Stack and returns as a Result object to the client.

Code to run the gRPC server

To run the gRPC server we need to:

Create a new instance of the gRPC struct and make it listen to one of the TCP ports at our localhost address. As a convention default port selected for gRPC is 9111.
To serve our StackMachine service over the gRPC server, we need to register the service with the newly created gRPC server.

For the development purpose, the basic insecure code to run the gRPC server should look like this.

var (
    port = flag.Int("port", 9111, "Port on which gRPC server should listen TCP conn.")
)

func main() {
    flag.Parse()
    lis, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    grpcServer := grpc.NewServer()
    machine.RegisterMachineServer(grpcServer, &server.MachineServer{})
    grpcServer.Serve(lis)
    log.Printf("Initializing gRPC server on port %d", *port)
}

We must consider strong TLS based security for our production environment. I’ll try planning to include an example of TLS implementation in this blog series.

Client

As we already know that the same machine/machine.proto file, which is our IDL (Interface Definition Language) is capable of generating interfaces for clients as well. One has to just implement those interfaces to communicate with the gRPC server.

With having a .proto either service provider can implement an SDK, or the consumer of the service itself can implement a client in the desired programming language.

Let's implement our version of a basic client code, which will call the Execute() method of the service. The client should look like this.

var (
    serverAddr = flag.String("server_addr", "localhost:9111", "The server address in the format of host:port")
)

func runExecute(client machine.MachineClient, instructions *machine.InstructionSet) {
    log.Printf("Executing %v", instructions)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    result, err := client.Execute(ctx, instructions)
    if err != nil {
        log.Fatalf("%v.Execute(_) = _, %v: ", client, err)
    }
    log.Println(result)
}

func main() {
    flag.Parse()
    var opts []grpc.DialOption
    opts = append(opts, grpc.WithInsecure())
    opts = append(opts, grpc.WithBlock())
    conn, err := grpc.Dial(*serverAddr, opts...)
    if err != nil {
        log.Fatalf("fail to dial: %v", err)
    }
    defer conn.Close()
    client := machine.NewMachineClient(conn)

    // try Execute()
    instructions := []*machine.Instruction{}
    instructions = append(instructions, &machine.Instruction{Operand: 5, Operator: "PUSH"})
    instructions = append(instructions, &machine.Instruction{Operand: 6, Operator: "PUSH"})
    instructions = append(instructions, &machine.Instruction{Operator: "MUL"})
    runExecute(client, &machine.InstructionSet{Instructions: instructions})
}

Test

Server

Let's write a unit test to validate our business logic of Execute() method.

Create a test file server/machine_test.go
Write the unit test, it should look like this.

Run the test file.

~/disk/E/workspace/grpc-eg-go
$ go test server/machine.go server/machine_test.go -v
=== RUN   TestExecute
--- PASS: TestExecute (0.00s)
PASS
ok      command-line-arguments    0.004s

Client

To test client-side code without the overhead of connecting to a real server, we'll use Mock. Mocking enables users to write light-weight unit tests to check functionalities on the client-side without invoking RPC calls to a server.

To write a unit test to validate client side business logic of calling the Execute() method:

Install golang/mock package
Generate mock for MachineClient
Create a test file mock/machine_mock_test.go
Write the unit test

As we are leveraging the golang/mock package, to install the package we need to run the following command:

~/disk/E/workspace/grpc-eg-go
$ go get github.com/golang/mock/mockgen@latest

To generate a mock of the MachineClient run the following command, the file should look like this.

~/disk/E/workspace/grpc-eg-go
$ mkdir mock_machine && cd mock_machine
$ mockgen github.com/toransahu/grpc-eg-go/machine MachineClient > machine_mock.go

Write the unit test, it should look like this.

Run the test file.

~/disk/E/workspace/grpc-eg-go
$ go test mock_machine/machine_mock.go  mock_machine/machine_mock_test.go -v
=== RUN   TestExecute
output:30
--- PASS: TestExecute (0.00s)
PASS
ok      command-line-arguments    0.004s

Run

Now we are assured through unit tests that the business logic of the server & client codes is working as expected, let’s try running the server and communicating to it via our client code.

Server

To turn on the server we need to run the previously created cmd/run_machine_server.go file.

~/disk/E/workspace/grpc-eg-go
$ go run cmd/run_machine_server.go

Client

Now, let’s run the client code client/machine.go.

~/disk/E/workspace/grpc-eg-go
$ go run client/machine.go
Executing instructions:<operator:"PUSH" operand:5 > instructions:<operator:"PUSH" operand:6 > instructions:<operator:"MUL" >
output:30

Hurray!!! It worked.

At the end of this blog, we’ve learned:

Importance of gRPC - What, Why, Where
How to install all the prerequisites
How to define an interface using protobuf
How to write gRPC server & client logic for Simple RPC
How to write and run the unit test for server & client logic
How to run the gRPC server and a client can communicate to it

The source code of this example is available at toransahu/grpc-eg-go.
You can also git checkout to this commit SHA to walk through the source code specific to this Part-1 of the blog series.

See you in the next part of this blog series.

DEV Community: Toran Sahu

Call for PyCon India 2020

Schedule & Venue

Participation

Volunteering

Proposals

PyCon India 2019

The Schedule

October 12-13

October 14-15

Day 1

Day 2

Why should you attend PyCon

A few days later

Links

Part-3: Building a basic microservice with bidirectional-streaming gRPC using Golang

Objective

Introduction

Implementing Bidirectional Streaming RPC

Update the protobuf

Generating the updated client and server interface Go code

Update the Server

Update the Client

Test

Server

Client

Run

Server

Client

Bonus

Part-2: Building a basic microservice with unidirectional-streaming gRPC using Golang

Objective

Introduction

Implementing Server Streaming RPC

Update the protobuf

Generating the updated client and server interface Go code

Update the Server

Update the Client

Test

Server

Client

Run

Server

Client

Implementing Client Streaming RPC

Update the protobuf

Generating the updated client and server interface Go code

Update the Server

Update the Client

Test

Server

Client

Run

Server

Client

Part-1: Building a basic microservice with gRPC using Golang

Introduction

Objective

What is gRPC?

Why use gRPC?

Where to use gRPC?

How to use gRPC?

Prerequisites

Go

gRPC

Protocol Buffers v3

Setting Project Structure

Defining the service

Generating client and server code

Server

Implementing Simple RPC

Code to run the gRPC server

Client

Test

Server

Client

Run

Server

Client