go-challenge/src/numbers/numbers.go

package main

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"net/url"
	"numbers/sort"
	"sync"
	"time"
)

// Helper struct for JSON decoding.
type Numbers struct {
	Numbers []int
}

// The maximum response time of the handlers.
var MaxResponseTime time.Duration = 500 * time.Millisecond

// The main entry point of the backend.
func main() {
	listenAddr := flag.String("http.addr", ":8090", "http listen address")
	flag.Parse()

	http.HandleFunc("/numbers", func(w http.ResponseWriter, r *http.Request) {
		numbersHandler(w, r)
	})

	log.Fatal(http.ListenAndServe(*listenAddr, nil))
}

// The main handler. The expected request is of the form:
//      GET /numbers?u=http://example.com/primes&u=http://foobar.com/fibo HTTP/1.0
// The parameter 'u' will be parsed and all urls will be fetched, which must
// return a valid JSON that looks like e.g.:
//     { "Numbers": [ 1, 2, 3, 5, 8, 13 ] }
//
// Then these lists are merged, deduplicated and sorted and the response will
// be a JSON of the same form.
// The handler is guaranteed to respond within a timeframe of 500ms. If
// URLs take too long to load or return garbage, they are skipped.
// If all URLs take too long to load or return garbage, an empty JSON list
// is returned.
func numbersHandler(w http.ResponseWriter, r *http.Request) {
	// timeout channel for the handler as a whole
	timeout := make(chan bool, 1)
	go func() {
		time.Sleep(MaxResponseTime)
		timeout <- true
	}()

	var rurl []string = r.URL.Query()["u"]
	// if no parameters, return 400
	if len(rurl) == 0 {
		w.Header().Set("Content-Type", "text/html")
		w.WriteHeader(http.StatusBadRequest)
		w.Write([]byte("Bad request: Missing 'u' parameters"))
		return
	}

	// Non-blocking input channel for URL results.
	// We will read as much as we can at once.
	inputChan := make(chan []int, len(rurl))
	var wg sync.WaitGroup

	// fetch all URLs asynchronously
	for i := range rurl {
		wg.Add(1)
		go func(url string) {
			defer wg.Done()
			n, e := getNumbers(url)

			if e == nil {
				if n != nil && len(n) > 0 {
					inputChan <- n
				} else {
					log.Printf("Received empty list of numbers from endpoint")
				}
			} else {
				log.Printf("Got an error: %s", e)
			}
		}(rurl[i])
	}
	// master routine closing the inputChan
	go func() {
		wg.Wait()
		close(inputChan)
	}()

	// channel for sorting process, so we can short-circuit in
	// case sorting takes too long
	sortChan := make(chan []int, 1)
	// aggregate numbers from URLs
	var numberBuffer []int = []int{}
	// these are actually sorted
	var sortedNumbers []int = []int{}

	// aggregate and sort loop,
	// breaks if all URLs have been processed or the timeout
	// has been reached
	done := false
	for done != true {
		select {
		case <-timeout:
			log.Printf("Waiting for URL took too long, finishing response anyway")
			finishResponse(w, sortedNumbers)
			return
		case res, more := <-inputChan:
			if more { // still URLs to fetch
				numberBuffer = append(numberBuffer, res...)
				// continue to aggregate numbers from the buffer
				continue
			} else { // all URLs fetched, sort and be done
				log.Printf("Nothing else to fetch")
				done = true
			}
		// non-blocking branch that sorts what we already have
		// we are not done here yet
		default:
			// only sort if we have new results
			if len(numberBuffer) == 0 {
				continue
			}
		}

		// sort fallthrough, either the inputChan is currently "empty"
		// or we fetched all URLs already
		go func(n []int) {
			res, err := sort.SortedAndDedup(timeout, n)
			if err != nil {
				return
			}
			sortChan <- res
		}(append(sortedNumbers, numberBuffer...))
		numberBuffer = []int{}

		select {
		case merged := <-sortChan:
			sortedNumbers = merged
		case <-timeout:
			log.Printf("Sorting took too long, finishing response anyway")
			finishResponse(w, sortedNumbers)
			return
		}
	}

	log.Printf("Result is complete, finishing response")
	finishResponse(w, sortedNumbers)
}

// Finalizes the JSON response with the given numbers. This always
// sends a 200 HTTP status code.
func finishResponse(w http.ResponseWriter, numbers []int) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(http.StatusOK)
	json.NewEncoder(w).Encode(map[string]interface{}{"Numbers": numbers})
}

// Gets the numbers from the given url.
// 'resp' is always nil if there was an error. Errors can
// be url parse errors, HTTP response errors, io errors from reading the
// body or json decoding errors.
func getNumbers(rawurl string) (resp []int, err error) {
	// validate url
	u_err := validateURL(rawurl)
	if u_err != nil {
		return nil, u_err
	}

	// retrieve response
	r, r_err := http.Get(rawurl)
	if r_err != nil {
		return nil, r_err
	}
	if r.StatusCode != 200 {
		return nil, fmt.Errorf("HTTP: Status code is not 200, but %d",
			r.StatusCode)
	}

	// open body
	defer r.Body.Close()
	body, b_err := ioutil.ReadAll(r.Body)
	if b_err != nil {
		return nil, b_err
	}

	// parse json
	return parseJson(body)
}

// Parse the given raw JSON bytes into a list of numbers. The JSON
// is expected to be of the form:
//     {"Numbers": [1,2,5]}
//
// Not particularly strict.
func parseJson(body []byte) (res []int, err error) {
	dec := json.NewDecoder(bytes.NewReader(body))
	var n Numbers
	j_err := dec.Decode(&n)
	if j_err != nil {
		return nil, j_err
	} else {
		if n.Numbers == nil {
			return nil, fmt.Errorf("JSON: missing key 'Numbers'")
		}
		return n.Numbers, nil
	}
}

// Validate the URL. The URL has to be syntactically valid,
// has to have a scheme of either 'https' or 'http' and a hostname.
// An error is returned on invalid URLs.
func validateURL(rawurl string) error {
	u, u_err := url.Parse(rawurl)
	if u_err != nil {
		return u_err
	}
	if u.Scheme != "https" && u.Scheme != "http" {
		return fmt.Errorf("URL: not a valid HTTP/HTTPS scheme in %s", rawurl)
	}
	if u.Host == "" {
		return fmt.Errorf("URL: not a valid host in %s", rawurl)
	}

	return nil
}