Sync async_nats work with master

Signed-off-by: Alex Ellis <alexellis2@gmail.com>

gateway/handlers/asyncreport.go

@@ -0,0 +1,23 @@
+package handlers
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/alexellis/faas/gateway/metrics"
+	"github.com/alexellis/faas/gateway/requests"
+)
+
+// MakeAsyncReport makes a handler for asynchronous invocations to report back into.
+func MakeAsyncReport(metrics metrics.MetricOptions) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		report := requests.AsyncReport{}
+		bytesOut, _ := ioutil.ReadAll(r.Body)
+		json.Unmarshal(bytesOut, &report)
+
+		trackInvocation(report.FunctionName, metrics, report.StatusCode)
+	}
+}

gateway/handlers/proxy.go

@@ -75,19 +75,6 @@ func MakeProxy(metrics metrics.MetricOptions, wildcard bool, client *client.Clie
 	}
 }
 
-func writeHead(service string, metrics metrics.MetricOptions, code int, w http.ResponseWriter) {
-	w.WriteHeader(code)
-
-	metrics.GatewayFunctionInvocation.With(prometheus.Labels{"function_name": service, "code": strconv.Itoa(code)}).Inc()
-
-	// metrics.GatewayFunctionInvocation.WithLabelValues(service).Add(1)
-}
-
-func trackTime(then time.Time, metrics metrics.MetricOptions, name string) {
-	since := time.Since(then)
-	metrics.GatewayFunctionsHistogram.WithLabelValues(name).Observe(since.Seconds())
-}
-
 func lookupInvoke(w http.ResponseWriter, r *http.Request, metrics metrics.MetricOptions, name string, c *client.Client, logger *logrus.Logger, proxyClient *http.Client) {
 	exists, err := lookupSwarmService(name, c)
 
@@ -197,3 +184,18 @@ func randomInt(min, max int) int {
 	rand.Seed(time.Now().Unix())
 	return rand.Intn(max-min) + min
 }
+
+func writeHead(service string, metrics metrics.MetricOptions, code int, w http.ResponseWriter) {
+	w.WriteHeader(code)
+
+	trackInvocation(service, metrics, code)
+}
+
+func trackInvocation(service string, metrics metrics.MetricOptions, code int) {
+	metrics.GatewayFunctionInvocation.With(prometheus.Labels{"function_name": service, "code": strconv.Itoa(code)}).Inc()
+}
+
+func trackTime(then time.Time, metrics metrics.MetricOptions, name string) {
+	since := time.Since(then)
+	metrics.GatewayFunctionsHistogram.WithLabelValues(name).Observe(since.Seconds())
+}

gateway/handlers/queueproxy.go

@@ -0,0 +1,43 @@
+package handlers
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/alexellis/faas/gateway/metrics"
+	"github.com/alexellis/faas/gateway/queue"
+	"github.com/docker/docker/client"
+	"github.com/gorilla/mux"
+)
+
+// MakeQueuedProxy accepts work onto a queue
+func MakeQueuedProxy(metrics metrics.MetricOptions, wildcard bool, client *client.Client, logger *logrus.Logger, canQueueRequests queue.CanQueueRequests) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		defer r.Body.Close()
+
+		body, err := ioutil.ReadAll(r.Body)
+
+		if err != nil {
+			w.WriteHeader(http.StatusBadRequest)
+			return
+		}
+
+		w.WriteHeader(http.StatusAccepted)
+		vars := mux.Vars(r)
+		name := vars["name"]
+		req := &queue.Request{
+			Function:    name,
+			Body:        body,
+			Method:      r.Method,
+			QueryString: r.URL.RawQuery,
+			Header:      r.Header,
+		}
+
+		err = canQueueRequests.Queue(req)
+		if err != nil {
+			fmt.Println(err)
+		}
+	}
+}