DEV Community: Promise Femi

Handing file upload with Go

Promise Femi — Thu, 25 May 2023 15:31:00 +0000

It has been one hell of a journey learning go, it been some-worth difficult as I was very used to the loose nature of JavaScript and PHP (languages I have been writing for about 3 years). GO's strict nature is something I have just been coming into terms with in the last few months and by all means, loving it. It is a great language.

The reason for this post is to try to explain file upload in GO through the eyes of an amateur, so pardon me if my writing is very verbose as i will be trying to explain this as i would to myself. so let's dig in.

We would be handling file uploading as part of the form body as well as handling multiple files.

First let's create the boilerplate, create a folder and create a new main.go file.

package main

func uploadHandler(w http.ResponseWriter, r *http.Request){

}

func main(){
        http.HandleFunc("/upload", uploadHandler)
        log.Fatal(http.ListenAndServe(":9000", nil))
}

next, let us create a simple index.html file where we would be making our submissions from

<!DOCTYPE html>
<html lang="en">

<head>
    <meta charset="UTF-8">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Go file upload</title>
</head>

<body>

    <form action="/upload" method="post">
        <input type="file" name="image" accept="image/*">
        <button type="submit">Submit</button>
    </form>

</body>

</html>

lets also serve the index.html file in main.go

package main

func uploadHandler(w http.ResponseWriter, r *http.Request){

}
func indexHandler(res http.ResponseWriter, r *http.Request) {
    res.Header().Add("Content-Type", "text/html")
    http.ServeFile(res, r, "index.html")
}

func main(){
http.HandleFunc("/", indexHandler)
        http.HandleFunc("/upload", uploadHandler)
        log.Fatal(http.ListenAndServe(":9000", nil))
}

I do recommend that when testing with the index.html file you add enctype="multipart/form-data" since browsers will not submit the file without this in the form tag and that would cause an error, or you could use postman until we get to parsing multipart forms.

next, we move forward to handling the form in the request, first, we must parse the form

func uploadHandler(w http.ResponseWriter, r *http.Request){
            // this neccessary else you will not be able to access other properties needed to get the file
// this method returns an err           
        err:=r.ParseForm()
// always handle error
     if err != nil{
// of course we should use a better error message
http.Error(w, "Unable to parse form", http.StatusInternalServerError)
return
}

}

the reason why parsing the form is that we need to call the r.FormFile() method and this method as well as the r.FormValue() methods are unavailable until the form is parsed.

in this case, we will be using the r.FormFile() method, this method returns three arguments:

First argument is the multipart.File type.
the second argument is the multipart.FIleHeader type these arguments contain the meta details of the file such as its name, size and the file itself."
the third and final argument is an error.

with that being said let jump in and take a look at the added lines of code

func uploadHandler(w http.ResponseWriter, r *http.Request) {

    // this neccessary else you will not be able to access other properties needed to get the file
    // this method returns an err           
            err:=r.ParseForm()
    // always handle error
     if err != nil{
            http.Error(w, "Unable to parse form",  http.StatusInternalServerError)
            return
        }

    file, fileHeader, err := r.FormFile("image") 
// get the file by entering its form name.
// handle errors as neccessary
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
//postpone closing the file until the function is complete
defer file.Close()

after the previous step, it is now time to create the file on the local server, we are going to be using the current time as the name of the file and extracting the file extension with the filepath.Ext() method.

func uploadHandler(w http.ResponseWriter, r *http.Request) {
/*
reduced for brevity
*/
// create file on the local server, filepath.Ext() will get the extension out of the filename
localfile, err := os.Create(time.Now().UTC().String()+filepath.Ext(fileHeader.Filename))
//handle errors as neccessary
if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

Now we can copy the uploaded file into the newly created file.

/*
reduced for brevity
*/
// create file on the local server, filepath.Ext() will get the extension out of the filename
localfile, err := os.Create(time.Now().UTC().String()+filepath.Ext(fileHeader.Filename))
//handle errors as neccessary
if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

// use the io.Copy() method to copy bytes to the localbyte from the uploaded file,
// this method returns the number of bytes copied and an error, we would be ignoring the bytes
_, err = io.Copy(localfile, file)
//handle errors as neccessary
if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

//after which we can simply respond back to the client
fmt.Fprint(w, "File upload successful")
}

and that's it, that is all that is needed to upload a file.

Parsing a multipart form

if the form submitted is a multipart form, parsing the form is slightly different from a normal form but only by one line.

//change this line
err:=r.ParseForm()
//to, r.ParseMultipartForm() takes an argument (maxMemory), maxMemory indicates how much of the file size should be kept in memory, while any extra will be kept in a temp folder on the disc
err:=r.ParseMultipartForm(300)

Uploading multiple files

Uploading multiple files is just as simple as uploading a single file with the major difference being how you handle getting the files and looping through them, lets take a look.

// create a new function for uploading multiple files
func uploadMultipleFiles(w http.ResponseWriter, r *http.Request){
        r.ParseMultipartForm(1024 * 1024 *10) //maxMemory if 10mb
}

let us also handle that in the main.go

func main() {

    http.HandleFunc("/", indexHandler)
    http.HandleFunc("/upload", uploadHandler)
    http.HandleFunc("/uploadmultiple", uploadMultipleFiles)

    http.ListenAndServe(":9000", nil)
}

after parsing the form, we would now be able to get the files with the r.MultipartForm.File["images"] slice.

func uploadMultipleFiles(w http.ResponseWriter, r *http.Request){
        r.ParseMultipartForm(1024 * 1024 *10) //maxMemory if 10mb

//we get the slice of files
files:= r.MultipartForm.File["images"]

}

we need to loop through the files to be able to upload each of the files just like we did before, with the only difference being that we would not be working with an actual multipart.File type but with a multipart.FileHeader type instead. but not to worry the multipart.FileHeader also contains the file itself.

func uploadMultipleFiles(w http.ResponseWriter, r *http.Request){
        r.ParseMultipartForm(1024 * 1024 *10) //maxMemory if 10mb

//we get the slice of files
files:= r.MultipartForm.File["images"]
// looping through the files to upload each file individually
    for _, file := range files {

        uploadFile, err := file.Open()
        // handle error as neccessary
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }

        // postpone file closure until the function is complete
        defer uploadFile.Close()

        //create a new localfile
        localfile, err := os.Create(time.Now().UTC().String() + filepath.Ext(file.Filename))
        //handle errors as neccessary
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }

        _, err = io.Copy(localfile, uploadFile)
        //handle errors as neccessary
        if err != nil {
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
    }

    //after which we can simply respond back to the client
    fmt.Fprint(w, "File upload successful")
}

Extras

Setting a maximum file upload size

in a case where you want to restrict the file upload size, we would simply equate the file size in bytes:

// add where neccessary in your code
const MAX_UPLOAD_SIZE int64 = 1024 * 1024 * 10 //maximum of 10mb per file
// if this was for a single file upload it would be fileHeader.Size, because it is the
// multipart.FileHeader type that has Size a property
if file.Size > MAX_UPLOAD_SIZE{
http.Error(w, "file sizes cannot be bigger than 10mb", http.StatusBadRequest)
            return
}

Tracking file upload progress

Following up with file upload so as to check if the file upload progress is a bit tricky but not complicated. First, we need to create a new struct type whose property would be the total size of the file (TotalBytesToRead) and how many bytes have been read (TotalBytesRead). the struct also needs to satisfy the io.Writer interface, OK let's see some code.

type Progress struct{
    TotalBytesToRead int64
    TotalBytesRead int64
}

// create the write method so satisfy the i0.Writer interface
func (p *Progress) Write(pb []byte) (int, error) {
    n := len(pb) //length of the file read so far
    // set error to nil since no error will be handled

    p.TotalBytesRead = int64(n)

    if p.TotalBytesRead == p.TotalBytesToRead {
        fmt.Println("Done")
        return n, nil
    }
//this will  be printed on your terminals
    fmt.Printf("File upload still in progress -- %d\n", p.TotalBytesRead)

            return n, nil
    }

// now in use with the file upload 

func uploadHandler(w http.ResponseWriter, r *http.Request) {

    // this neccessary else you will not be able to access other properties needed to get the file
    // this method returns an err           
            err:=r.ParseForm()
    // always handle error
     if err != nil{
            http.Error(w, "Unable to parse form", http.StatusInternalServerError)
            return  
        }

    file, fileHeader, err := r.FormFile("image") 
// get the file by entering its form name.
// handle errors as neccessary
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
//postpone closing the file until the function is complete
defer file.Close()

//create a new localfile 
localfile, err := os.Create(time.Now().UTC().String()+filepath.Ext(file.Filename))
//handle errors as neccessary
if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

// Create a new instance of progress
progress := &Progress{
TotalBytesToRead : fileHeader.Size,
}

//instead of coppying directly we would use the io.TeeReader() method
// this method will allow us to write the amount of bytes read into the progress.Write method
_, err := io.Copy(localfile, io.TeeReader(file, progress))
//handle errors as neccessary
if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

}

and that is it, that is all that is needed to track file upload progress.

at this point, I hope you are now a master in the arts of file uploading and all of its caveats. thank you for reading and i hope this was helpful. please follow me.

Complete code can be found here

Data Structures: Singly Linked List

Promise Femi — Thu, 25 May 2023 15:28:35 +0000

Original article is available on my website: https://promisefemi.vercel.app/blog/data-structures-singly-linked-list

The illusive data structures, in this series i would be explaining individual data structures and their respective implementation in the Go programming language.

Linked list are a very simple data structure and they are often compared to arrays, but there are distinct differences not just in how they are implemented but in how they work under the hood.

The values of arrays are stored next to each other in memory, making it very easy to find a specific value by reaching out directly to the index.

This is a visual representation of an array - 200,201, 202 … represents space in memory

Image source: https://www.geeksforgeeks.org

A linked list on the other hand are arranged in nodes where each node points to the next node in memory, the values in memory may not be close to each other..

Image source: https://www.geeksforgeeks.org

Each node from the head (the first node) points to the next node and that node points to the next and so on.

This is the point where most people begin to ask why they should use a linked list instead of just using an array, saving themselves the stress. But linked list are particularly useful for storing ordered data and other forms of linked list such as the doubly linked list are useful for things like undo/redo, music playlists to name a few

Alright enough talk, show me the code

Implementing a Singly Linked list

Implementing a singly linked list is very simple ( trust me 😉)

First we need to create the structure of the linked list

package main

type linkedList struct {
// value can be whatever datatype fits your particular scenario
    value string
    next  *linkedList
}

in the example above we created a new struct type with a value and next field, the value field is the value of the current node and the next field points to the next node on the linked list. Let us initialize a new linkedList

func main(){
    list := new(linkedList)
}

That’s it ( well not particularly ), that is all you need for the structure of a linked list. Now let’s talk about inserting a new node into a linked list .

Inserting a new node

To insert a new node into the linked list, we have to move through the linked list until we find the last node, then we just add the new node to the last position. if that confuses you let me explain in code.

func (node *linkedList) insert(text string){
        // first we create a temporary value for our new node
        tempNode :=  &linkedList{value: text, next: nil}
        // remember that the next pointer in the last node is always nil until it points to another node
        if node == nil{
        // if this is a head node and it is nill, make the current tempNode the head
            node = tempNode
        }else{          
                currentNode :=  node
                // go through every node until we get to the last node,             
                for currentNode.next != nil{
                    // check if it is the last node, if not move to the next 
                    currentNode =  currentNode.next
                }
            // when we get to the last node, place the new node we are creating
            currentNode.next = tempNode
        }
}

Now when we go back to our main() we would call the insert method to add new nodes to our linked list

func main(){
    list := new(linkedList)
    list.insert("Apple")
    list.insert("Oranges")
    list.insert("Strawberries")
}

At this point you are probably like “ Yeah we’ve been doing all this things but how do we even know it’s adding anything “, Don’t worry i got you.

Displaying values

Let’s add a method that loops through each of the nodes to print out their values.

func (node *linkedList) print(){
// start from the first node
    currentNode := node
    // loop through each node, making sure it is not nil
    for currentNode != nil{
        //print the value out
        fmt.Printf("%s -> ", currentNode.value)
        //move to the next node
        currentNode = currentNode.next
    }
}

All together now

package main

type linkedList struct {
// value can be whatever datatype fits your particular scenario
    value string
    next  *linkedList
}

func (node *linkedList) insert(text string){
        // first we create a temporary value for our new node
        tempNode :=  &linkedList{value: text, next: nil}
        // remember that the next pointer in the last node is always nil until it points to another node
        if node == nil{
        // if this is a head node and it is nill, make the current tempNode the head
            node = tempNode
        }else{          
                currentNode :=  node
                // go through every node until we get to the last node,             
                for currentNode.next != nil{
                    // check if it is the last node, if not move to the next 
                    currentNode =  currentNode.next
                }
            // when we get to the last node, place the new node we are creating
            currentNode.next = tempNode
        }
} 

func (node *linkedList) print(){
// start from the first node
    currentNode := node
    // loop through each node, making sure it is not nil
    for currentNode != nil{
        //print the value out
        fmt.Printf("%s -> ", currentNode.value)
        //move to the next node
        currentNode = currentNode.next
    }
}

func main(){
    list := new(linkedList)
    list.insert("Apple")
    list.insert("Oranges")
    list.insert("Strawberries")

    list.print()
}

With all of that intact, let’s make it more interesting by allowing our users to select between adding new nodes to the linked list and printing the entire linked list

package main

import (
    "fmt"
    "log"
    "os"
)

type linkedList struct {
    // value can be whatever datatype fits your particular scenario
    value string
    next  *linkedList
}

func (node *linkedList) insert(text string) {
    // first we create a temporary value for our new node
    tempNode := &linkedList{value: text, next: nil}
    // remember that the next pointer in the last node is always nil until it points to another node
    if node == nil {
        // if this is a head node and it is nill, make the current tempNode the head
        node = tempNode
    } else {
        currentNode := node
        // go through every node until we get to the last node,
        for currentNode.next != nil {
            // check if it is the last node, if not move to the next
            currentNode = currentNode.next
        }
        // when we get to the last node, place the new node we are creating
        currentNode.next = tempNode
    }
}

func (node *linkedList) print() {
    // start from the first node
    currentNode := node
    // loop through each node, making sure it is not nil
    for currentNode != nil {
        //print the value out
        fmt.Printf("%s -> ", currentNode.value)
        //move to the next node
        currentNode = currentNode.next
    }
    fmt.Println("")
}

func main() {
    list := new(linkedList)

    for {
        var selected string
        fmt.Println("(1) enter a new item")
        fmt.Println("(2) print list")
        _, err := fmt.Scanln(&selected)
        if err != nil {
            log.Fatalln(err)
        }

        switch selected {
        case "1":
            var item string
            _, err := fmt.Scanln(&item)
            if err != nil {
                log.Fatalln(err)
            }
            list.insert(item)
        case "2":
            list.print()
        default:
            os.Exit(0)
        }
    }
}

Hurray!!, That’s it (no for real that’s it) we have a linked list that would be proud enough to carry its shoulders up, but i will be leaving you with the task of removing a node from the linked list by simply passing a value to a method ( call it homework ).

The code in this article is available here

Thank’s for reading, hope to see you again next time, and please share.

gRPC Client Connection Pooling

Promise Femi — Thu, 25 May 2023 15:03:25 +0000

Original article was published on my website: https://promisefemi.vercel.app/blog/grpc-client-connection-poolin

Recently I was working on a daemon for consuming MQTT messages sent from thousands of IOT devices. One of the requirements was that the daemon needed to be able to handle about 15k messages per second (I know right 😮 ).

Without going into too much detail (that would be for another blog) we decided to cut the round trip of our MySQL database completely and introduce a gRPC server that would act as a buffer between our daemon and the database. By doing that our daemon would simply drop messages off to the gRPC server and the server would queue and write to the database on its own time.

Building the gRPC server was the easy part the hard part is managing the needs of the daemons, a number of goroutines would need to send messages to the gRPC server, but using the same client connection would significantly affect performance since new requests would have to wait for the connection to finish whatever task is currently being handled, just imagine 14K+ messages waiting for one connection to become free. At this point it was clear using one connection was a no-go and creating new connections is out of the picture (creating new connections is very expensive here, so we decided to use a connection pool.

A lot of the ideas here were inspired by This Blog, i strongly recommend it.

If you don’t know what a connection pool is I strongly advise that you read the blog I linked above.

Jumping In

If you don’t want to go through everything, i have already created a go package that can be used right away. https://github.com/promisefemi/grpc-client-pool

Before anything else, let's talk about how to create new gRPC client connections

conn, err := grpc.Dial(":9000", grpc.WithTransportCredentials(insecure.NewCredential()), grpc.WithBlock())
if err != nil{
    // Handle error
}
// create a new client and use connection

or with context

ctx, cancelCtx := context.WithTimeout(context.Background(), 3 * time.Second)
conn, err := grpc.DialContext(ctx,":9000", grpc.WithTransportCredentials(insecure.NewCredential()), grpc.WithBlock())
if err != nil{
    // Handle error
}
// create a new client and use connection

Pool Structure

At its core, a connection pool contains three important properties maxOpenConnection, maxIdleConnection and idleConnections

maxOpenConnection - indicates how many connections we are allowed to open before we have to start queue connection requests.

maxIdleConnection - how many connections are we allowed to keep before we start closing connections?

idleConnections - a list of connections that can be reused.

There are also two important methodsGet and put these methods are responsible for getting and caching connections. Altogether these properties and methods are responsible for managing our connection pool.

type ClientPool struct {
    mu sync.Mutex //mutex to protect the idleConnections and numOfOpenConnection properties

    address             string // gRPC server address
    configOptions       []grpc.DialOption // gRPC dial configurations
    maxOpenConnection   int // maximum number of open connections allowed
    maxIdleConnection   int // maximum number of idle connections allwed
    idleConnections     map[string]*ClientCon // list of idle connections
    numOfOpenConnection int // number of currently open connections
}

ClientCon is a custom type, so let's create that as well

type ClientCon struct {
    id   string
    pool *ClientPool
    Conn *grpc.ClientConn
}

Now let us proceed with our put and get methods

Put method

The responsibility of the put method is very simple, it simply takes a connection and returns it back to the list of idle connections (the pool) or closes it if there is no room.

func (cp *ClientPool) put(conn *ClientCon) {
    //Lock the pool and defer the unlock 
    cp.mu.Lock()
    defer cp.mu.Unlock()
    // check to see if we can return the connection back to the idleConnection
    if cp.maxIdleConnection >= len(cp.idleConnections) {
        cp.idleConnections[conn.id] = conn
    } else {
    // if the number of idle connections are more than allowed, close the connection
        cp.numOfOpenConnection--
        _ = conn.Conn.Close()
    }
}

and that’s about it, the put method is done.

Get method

The get method is a little bit different, the get method has to handle three different scenarios:

Return a connection from the list of idle connections if there is any available
Queue connection request if we have surpassed the number of connections we can create
Create a new connection

Let's see how this translates into code.

func (cp *ClientPool) Get() (*ClientCon, error) {
    //Lock the pool
    cp.mu.Lock()

// check if we have any connection available 
    if len(cp.idleConnections) > 0 {
  // return the first connection available, and remove from the list of idleConnection
        for _, val := range cp.idleConnections {
            delete(cp.idleConnections, val.id)
            // increment the number of open connections
            cp.numOfOpenConnection++
            cp.mu.Unlock()
            return val, nil
        }
    }
...

For the first step, we check if we have any idle connections and if we do return the first connection, and increment the list of open connections.

If there are no available connections we move to the next step.

Before we start creating new connections, its important to know if we have not surpassed the number of open connections we are allowed to create, and if we have we would need to queue the connection request and resolve it at a later time.

Go channels are coming to the rescue here. But before we do that we need to introduce a new type that’s going to act as the channel type.

The queueChan struct contains two properties a, this channel is used to receive a ClientCon request once it is fulfilled and an errorChan a channel that receives an error if a connection request could not be fulfilled.

// A queue struct ( The struct is not a channel i just named it that way)
type queueChan struct {
    connectionChan chan *ClientCon
    errorChan      chan error
}

and let's add an update to the ClientPool to contain the channel that would act as the queue.

type ClientPool struct {
    ...
    connectionQueue     chan *queueChan // channel for queuing connection requests when there are no idle connections left. 
}

Back in the get method, we need to create a new queue request for a connection, pass that request to the pool connection queue then use a select to block until the request is resolved

...
// Check if we are allowed to open new connections 
if cp.maxOpenConnection > 0 && cp.numOfOpenConnection >= cp.maxOpenConnection {
    // create a new queue request 
        queueRequest := &queueChan{
            connectionChan: make(chan *ClientCon),
            errorChan:      make(chan error),
        }
        // pass request into connection queue
        cp.connectionQueue <- queueRequest

        // block with a select until the request is fuffiled or an error occurs instead
        select {
        case conn := <-queueRequest.connectionChan:
            cp.numOfOpenConnection++
            cp.mu.Unlock()
            return conn, nil
        case err := <-queueRequest.errorChan:
            cp.mu.Unlock()
            return nil, err
        }
    } 
...

With that we have successfully queued the connection request and when an idle connection is available the request would be resolved. We would come back to how we intend on resolving requests from the connection queue.

The last part of the get method is very simple, at this point, there is no idle connection and we can create a new connection (instead of queuing it).

conn, err := cp.openConnection()
    if err != nil {
        return nil, err
    }

    cp.numOfOpenConnection++
    cp.mu.Unlock()
    return conn, nil
}

func (cp *ClientPool) openConnection() (*ClientCon, error) {
    // Dial a new grpc client connection or use DialContext if that works for you
    newConn, err := grpc.Dial(cp.address, cp.configOptions...)

    if err != nil {
        return nil, err
    }
    // create a new ClientCon
    return &ClientCon{
        id:   fmt.Sprintf("%v", time.Now().Unix()),
        pool: cp,
        Conn: newConn,
    }, nil
}

and all together

// A queue struct ( The struct is not a channel i just named it that way)
type queueChan struct {
    connectionChan chan *ClientCon
    errorChan      chan error
} 

func (cp *ClientPool) Get() (*ClientCon, error) {
    //Lock the pool
    cp.mu.Lock()

// check if we have any connection available 
    if len(cp.idleConnections) > 0 {
  // return the first connection available, and remove from the list of idleConnection
        for _, val := range cp.idleConnections {
            delete(cp.idleConnections, val.id)
            // increment the number of open connections
            cp.numOfOpenConnection++
            cp.mu.Unlock()
            return val, nil
        }
    }
    // Check if we are allowed to open new connections 
    if cp.maxOpenConnection > 0 && cp.numOfOpenConnection >= cp.maxOpenConnection {
        // create a new queue request 
            queueRequest := &queueChan{
                connectionChan: make(chan *ClientCon),
                errorChan:      make(chan error),
            }
            // pass request into connection queue
            cp.connectionQueue <- queueRequest

            // block with a select until the request is fuffiled or an error occurs instead
            select {
            case conn := <-queueRequest.connectionChan:
                cp.numOfOpenConnection++
                cp.mu.Unlock()
                return conn, nil
            case err := <-queueRequest.errorChan:
                cp.mu.Unlock()
                return nil, err
            }
        } 
        conn, err := cp.openConnection()
        if err != nil {
                return nil, err
        }

        cp.numOfOpenConnection++
        cp.mu.Unlock()
        return conn, nil
}

func (cp *ClientPool) openConnection() (*ClientCon, error) {
    // Dial a new grpc client connection or use DialContext if that works for you
    newConn, err := grpc.Dial(cp.address, cp.configOptions...)

    if err != nil {
        return nil, err
    }
    // create a new Clientcon
    return &ClientCon{
        id:   fmt.Sprintf("%v", time.Now().Unix()),
        pool: cp,
        Conn: newConn,
    }, nil
}

And with all that we are almost done. Remember when we queued connection requests? well it's time to resolve those requests.

To do that we need to add a method that would run on a separate goroutine, constantly waiting for queued connection requests and the chance to resolve them. let's call this method handleConnectionQueue

Handling Connection Queue

There are three basic things we need to care about when handling connection queue requests:

Watching for new requests in the queue
Checking for a newly idle connection
Creating a connection if there are no idle connections and if we are allowed to do so.

As you may have noticed there are a lot of similarities with the Get method, but something to take note of is that we are not returning an error immediately after a connection request fails instead we will continue to retry for a given time frame before deciding that the request failed.

Let us begin

func (cp *ClientPool) handleConnectionQueue() {
    for rq := range cp.connectionQueue {

        var (
            hasTimedOut  = false
            hasCompleted = false
            timeout      = time.After(time.Duration(3) * time.Second)
        )
        //continually try to get/create a connection until timeout or connection completed
        for {

            if hasCompleted || hasTimedOut {
                break
            }
            //continually check for timeout or try to get/create a connection
            select {
            case <-timeout:
                hasTimedOut = true
                rq.errorChan <- ErrConnectionWaitTimeout
            default:
            ...

And within the default: we would be doing two things, trying to get an idle connection or trying to create a new connection

                ...
                //  first check if an idle connection is available
                cp.mu.Lock()
                numberOfIdleConnections := len(cp.idleConnections)
                if numberOfIdleConnections > 0 {
                    for _, val := range cp.idleConnections {
                        delete(cp.idleConnections, val.id)
                        cp.mu.Unlock()
                        rq.connectionChan <- val
                        hasCompleted = true
                        break
                    }
                } else if cp.maxOpenConnection > 0 && cp.maxOpenConnection > cp.numOfOpenConnection {
                    //check if pool has not exceeded number of allowed open connections
                    // increase numberOfConnection hoping new connection would be created
                    // unlock mutext to free up resources for other connection, since creating new connections could take a while
                    cp.numOfOpenConnection++
                    cp.mu.Unlock()

                    conn, err := cp.openConnection()
                    //ignoring error because the only error we care about is the timeout, we continue to retry
                    cp.mu.Lock()
                    cp.numOfOpenConnection--
                    cp.mu.Unlock()
                    if err == nil {
                        rq.connectionChan <- conn
                        hasCompleted = true
                    }

                } else {
                    //unlock pool and restart
                    cp.mu.Unlock()
                }
            }
        }
    }
}

All together

func (cp *ClientPool) handleConnectionQueue() {
    for rq := range cp.connectionQueue {

        var (
            hasTimedOut  = false
            hasCompleted = false
            timeout      = time.After(time.Duration(3) * time.Second)
        )
        //continually try to get/create a connection until timeout or connection completed
        for {

            if hasCompleted || hasTimedOut {
                break
            }
            //continually check for timeout or try to get/create a connection
            select {
            case <-timeout:
                hasTimedOut = true
                rq.errorChan <- ErrConnectionWaitTimeout
            default://  first check if a idle connection is available
                cp.mu.Lock()
                numberOfIdleConnections := len(cp.idleConnections)
                if numberOfIdleConnections > 0 {
                    for _, val := range cp.idleConnections {
                        delete(cp.idleConnections, val.id)
                        cp.mu.Unlock()
                        rq.connectionChan <- val
                        hasCompleted = true
                        break
                    }
                } else if cp.maxOpenConnection > 0 && cp.maxOpenConnection > cp.numOfOpenConnection {
                    //check if pool has not exceeded number of allowed open connections
                    // increase numberOfConnection hoping new connection would be created
                    // unlock mutext to free up resources for other connection, since creating new connections could take a while
                    cp.numOfOpenConnection++
                    cp.mu.Unlock()

                    conn, err := cp.openConnection()
                    //ignoring error because the only error we care about is the timeout, we continue to retry
                    cp.mu.Lock()
                    cp.numOfOpenConnection--
                    cp.mu.Unlock()
                    if err == nil {
                        rq.connectionChan <- conn
                        hasCompleted = true
                    }

                } else {
                    //unlock pool and restart
                    cp.mu.Unlock()
                }
            }
        }
    }
}

That is most of what we need to build a functional pool.

We still need to write a method to release a connection when you're done using it, for ease of usage we can bind that method to the ClientCon . All the method does is call the ClientPool put method, as i have stated above all the put method does is return a connection back to the list of idle connections or close it completely.

func (c *ClientCon) Release() {
    c.pool.put(c)
}

And for the last part, we want to write a function that helps us create a new pool, configure it and start the handleConnectionQueue goroutine

type PoolConfig struct {
    MaxOpenConnection     int // number of open Connections
    MaxIdleConnection     int // number of idle connections
    ConnectionQueueLength int //buffer capacity of the queue channel
    Address               string // gRPC server address
    ConfigOptions         []grpc.DialOption // gRPC connection options
}

func NewClientPool(config *PoolConfig) *ClientPool {
    clientPool := &ClientPool{
        mu:                  sync.Mutex{},
        address:             config.Address,
        configOptions:       config.ConfigOptions,
        maxOpenConnection:   config.MaxOpenConnection,
        maxIdleConnection:   config.MaxOpenConnection,
        numOfOpenConnection: 0,
        connectionQueue:     make(chan *queueChan, config.ConnectionQueueLength),
        idleConnections:     make(map[string]*ClientCon, 0),
    }
// start goroutine
    go clientPool.handleConnectionQueue()
    return clientPool
}

For real now, that's it we have successfully created a gRPC client connection pool, to create a new pool we would simply

...
//Create a new pool config
poolConfig := &pool.PoolConfig{
        MaxOpenConnection:     10,
        MaxIdleConnection:     10,
        ConnectionQueueLength: 10000,
        Address:               ":9000",
        ConfigOptions: []grpc.DialOption{
            grpc.WithTransportCredentials(insecure.NewCredentials()),
            grpc.WithBlock(),
        },
    }

// create a new client pool
    connPool := pool.NewClientPool(poolConfig)

//get a new ClientConn
        client, err := connPool.Get()
        if err != nil {
            log.Fatalf("%s", err)
        }

// Setup a new grpc client
// send a request using the conn
        userMessage := &proto.UserMessage{
            FirstName: "Promise",
            LastName:  "Femi",
            Email:     "",
        }

        uc := proto.NewUserClient(client.Conn)
        response, err := uc.Set(context.Background(), userMessage)
        if err != nil {
            fmt.Printf("error unable to set user -- %s -- %d\n", err, connPool.GetNumberOfOpenConnections())
        } else {
            fmt.Printf("%+v -- %d \n", response, connPool.GetNumberOfOpenConnections())
        }
// remember to release connections.
        client.Release()

We are at the end of a great journey 🙇🏾. Thank you for reading my article, and you can reach out to me if you have any feedback ✌🏾.

I have already created a package that implements all the features outlined in this article @ https://github.com/promise/grpc-client-pool

I Learned React now I need practice

Promise Femi — Thu, 29 Aug 2019 19:32:29 +0000

Hi, I am a new developer (the new guy) and I just learned react and redux. and of course to really get good at something (especially codding) it is essential to do a lot of practice and i have built more todos and cart than React tutorials on youtube, it's now borring***.

I need something more challenging, hopefully including Redux and React-Router.

I just need some kind of list of cool stuff to build.

Thanks a lot.