Golang Pipeline in Practise

I have a request to perform some Google API search for some keywords and save the result (the URLs) into text files. It's a good chance to use Go’s concurrency pattern to assembly the pipeline, therefore, parallelizing the IO and CPU.

Google Search API

Given a keyword, say a category, the following function calls Google search API using go-resty, then picks up the respective Links from the result using the gjson library.

func SearchGoogle(category string, count int) ([]SearchResult, error) {
 results := []SearchResult{}
 // ... skip some initial assignment
 for i := 0; i < pages; i++ {
  resp, err := client.R().
   SetQueryParams(map[string]string{
    "cx":  cx,
    "key": key,
    "q":   category,
    "start": fmt.Sprintf("%d", i*10+1)
   }).
   SetHeader("Accept", "application/json").
   Get("https://www.googleapis.com/customsearch/v1")
  //... skip the error handling part 
  content := resp.Body()
  if resp.StatusCode() != 200 {
   logrus.Errorf("staus code is non 200, code:%v", resp.StatusCode())
   return results, err
  }
  gjson.GetBytes(content, "items").ForEach(func(_, elm gjson.Result) bool {
   counter += 1
   title := elm.Get("title").String()
   link := elm.Get("link").String()
   snippet := elm.Get("snippet").String()
   
   results = append(results, SearchResult{
    Category: category,
    Title:    title,
    Link:     link,
    Snippet:  snippet,
   })
   return true //keep looping
  })
 }
 return results, nil
}

We read the keyword to search into a slice of string. Let set up the golang pipeline with channels.

Data Feeding

To start the pipeline, take in a slice of string and output a channel of string.

func Feed(list []string) chan string {
 out := make(chan string)
 go func() {
  for _, v := range list {
   out <- v
  }
  close(out)
 }()
 return out
}

First, create the “out” chan. Then create a goroutine to loop through the keyword list, send it to the out channel. Once the looping is done, close the out channel to inform the receiver end, it’s closed.

Notice the sending to the channel is blocking unless the pipeline is connected and moving. That’s why we…