HTTP Multiplexor in Go and Kotlin

multiplexor.go

package main

import (
	"net/http"
	"time"
	"io/ioutil"
	"fmt"
	"net/url"
)

const (
	googleEndpoint     = "https://www.googleapis.com/customsearch/v1?q=%s"
	bingEndpoint       = "https://api.cognitive.microsoft.com/bing/v7.0/search?q=%s"
	duckduckgoEndpoint = "https://api.duckduckgo.com/?q=%s&format=json&pretty=1"
	baiduEndpoint      = "https://www.baidu.com/s?wd=%s"
)

var (
	providers = map[string]string {
		"Google":     googleEndpoint,
		"Bing":       bingEndpoint,
		"DuckDuckGo": duckduckgoEndpoint,
		"Baidu":      baiduEndpoint,
	}
)

type HttpRequest struct {

	Provider string

	Endpoint string

}

type HttpResponse struct {

	Provider string

	Response string

	Success bool

	Start int64

	Finish int64

}

func timestamp() int64 {
	return time.Now().UnixNano() / int64(time.Millisecond)
}

func Query(query string) <-chan HttpResponse {

	encoded := url.QueryEscape(query)

	var requests []HttpRequest

	for key, value := range providers {
		requests = append(requests, HttpRequest{ Provider: key, Endpoint: fmt.Sprintf(value, encoded) })
	}

	client := http.Client{}

	return multiplex(client, requests...)
}

//receives an HTTP Client instance and a variadic list of HTTP Requests to execute asynchronously;
//multiplexes the output of each request (it's response) into a singular channel
func multiplex(client http.Client, requests ...HttpRequest) <-chan HttpResponse {

	channel := make(chan HttpResponse)

	for _, request := range requests {

		//fire and forget: launches a goroutine that is executed sometime in the future
		//and sends it's output into the channel returned from the enclosing scope
		go func(req HttpRequest) {

			start := timestamp()

			fmt.Printf("Executing query: %s at %d\n", req.Endpoint, start)

			if resp, err := client.Get(req.Endpoint); err == nil {

				if buf, err := ioutil.ReadAll(resp.Body); err == nil {

					channel <- HttpResponse{
						Provider: req.Provider,
						Response: string(buf),
						Success: true,
						Start: start,
						Finish: timestamp(),
					}

					return

				}

			}

			channel <- HttpResponse{
				Provider: req.Provider,
				Success: false,
				Start: start,
				Finish: timestamp(),
			}

		}(request)

	}

	return channel
}

func main() {

	//suspend execution after a 5 seconds have elapsed, allowing the program to exit
	timeout := time.After(5 * time.Second)
	results := Query("gmail")

	for {
		select {
			case <-timeout: //on timeout, break out of the infinite loop
				return

			case result := <-results:
				fmt.Printf("Took %s %dms to respond. Success: %v \n", result.Provider, result.Finish - result.Start, result.Success)
				fmt.Println(result.Response)

		}
	}

}

multiplexor.kt

package http

import kotlinx.coroutines.experimental.channels.Channel
import kotlinx.coroutines.experimental.channels.ReceiveChannel
import kotlinx.coroutines.experimental.launch
import kotlinx.coroutines.experimental.runBlocking
import kotlinx.coroutines.experimental.withTimeoutOrNull
import okhttp3.OkHttpClient
import okhttp3.Request
import java.net.URLEncoder
import java.util.concurrent.TimeUnit

private const val GOOGLE_ENDPOINT = "https://www.googleapis.com/customsearch/v1?q=%s"
private const val BING_ENDPOINT = "https://api.cognitive.microsoft.com/bing/v7.0/search?q=%s"
private const val DUCKDUCKGO_ENDPOINT = "https://api.duckduckgo.com/?q=%s&format=json&pretty=1"
private const val BAIDU_ENDPOINT = "https://www.baidu.com/s?wd=%s"

private val PROVIDERS = mapOf(
        "Google" to GOOGLE_ENDPOINT,
        "Bing" to BING_ENDPOINT,
        "DuckDuckGo" to DUCKDUCKGO_ENDPOINT,
        "Baidu" to BAIDU_ENDPOINT
)


data class HttpRequest(
        val provider: String,
        val endpoint: String
)

data class HttpResponse(
        val provider: String,
        val response: String,
        val success: Boolean,
        val start: Long,
        val finish: Long
) {

    fun elapsed() = finish - start

}

fun timestamp() = System.currentTimeMillis()

fun query(query: String): ReceiveChannel<HttpResponse> {

    val encoded = URLEncoder.encode(query, "UTF-8")

    val requests = PROVIDERS.map {
        HttpRequest(
                provider = it.key,
                endpoint = String.format(it.value, encoded)
        )
    }.toTypedArray()

    val client = OkHttpClient()

    return multiplex(client, requests = *requests)
}

//receives an HTTP Client instance and a variadic list of HTTP Requests to execute asynchronously;
//multiplexes the output of each request (it's response) into a singular channel
fun multiplex(client: OkHttpClient, vararg requests: HttpRequest): ReceiveChannel<HttpResponse> {

    //extension function
    fun OkHttpClient.get(start: Long, request: HttpRequest): HttpResponse {
        val req = Request.Builder()
                .url(request.endpoint)
                .build()

        val resp = this.newCall(req).execute()

        return HttpResponse(
                provider = request.provider,
                response = resp.body()!!.string(),
                success = true,
                start = start,
                finish = timestamp()
        )
    }

    val channel = Channel<HttpResponse>()

    for (request in requests) {

        //fire and forget: launches a coroutine that is executed sometime in the future
        //and sends it's output into the channel returned from the enclosing scope
        launch {

            val start = timestamp()

            println("Executing query: ${request.endpoint} at ${start}")

            val response = try {
                client.get(start, request)
            } catch (e: Throwable) {
                HttpResponse(
                        provider = request.provider,
                        response = "",
                        success = false,
                        start = start,
                        finish = timestamp()
                )
            }

            channel.send(response)

        }

    }

    return channel
}

fun main(args: Array<String>) = runBlocking {

    val results = query("gmail")

    //suspend execution after 5 seconds have elapsed, allowing the program to exit
    withTimeoutOrNull(5, TimeUnit.SECONDS) {

        while (true) { //this code is suspended when the enclosing CourtineScope is suspended (i.e, stops executing)

            val result = results.receive()

            println("Took ${result.provider} ${result.elapsed()}ms to respond. Success: ${result.success}")
            println(result.response)
        }

    }

    //semantics: main must return an object of type Unit (void in Java).
    //the last value in the scope of a closure is its return value.
    Unit
}