Refactor async runner and its reference and add unit tests

api/mqs/memory.go

@@ -131,6 +131,7 @@ func (mq *MemoryMQ) Push(job *models.Task) (*models.Task, error) {
 	// Push the work onto the queue.
 	return mq.pushForce(job)
 }
+
 func (mq *MemoryMQ) pushTimeout(job *models.Task) error {
 
 	ji := &TaskItem{

api/runner/async_runner.go

@@ -3,6 +3,7 @@ package runner
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -15,44 +16,32 @@ import (
 	"github.com/iron-io/functions/api/models"
 )
 
-func RunAsyncRunners(mqAdr string) {
-
-	url := fmt.Sprintf("http://%s/tasks", mqAdr)
-
-	logAndWait := func(err error) {
-		log.WithError(err)
-		time.Sleep(1 * time.Second)
-	}
-
-	for {
+func getTask(url string) (*models.Task, error) {
 	resp, err := http.Get(url)
 	if err != nil {
-			logAndWait(err)
-			continue
+		return nil, err
 	}
 
 	body, err := ioutil.ReadAll(resp.Body)
 	if err != nil {
-			logAndWait(err)
-			continue
+		return nil, err
 	}
 
 	var task models.Task
 
 	if err := json.Unmarshal(body, &task); err != nil {
-			logAndWait(err)
-			continue
+		return nil, err
 	}
 
 	if task.ID == "" {
-			time.Sleep(1 * time.Second)
-			continue
+		return nil, errors.New("Invalid Task: ID empty")
 	}
+	return &task, nil
+}
 
-		log.Info("Picked up task:", task.ID)
+func getCfg(task *models.Task) *Config {
 	var stdout bytes.Buffer                                           // TODO: should limit the size of this, error if gets too big. akin to: https://golang.org/pkg/io/#LimitReader
 	stderr := NewFuncLogger(task.RouteName, "", *task.Image, task.ID) // TODO: missing path here, how do i get that?
-
 	if task.Timeout == nil {
 		timeout := int32(30)
 		task.Timeout = &timeout
@@ -66,33 +55,70 @@ func RunAsyncRunners(mqAdr string) {
 		Stderr:  stderr,
 		Env:     task.EnvVars,
 	}
+	return cfg
+}
 
-		metricLogger := NewMetricLogger()
-
-		rnr, err := New(metricLogger)
+func deleteTask(url string, task *models.Task) error {
+	// Unmarshal task to be sent over as a json
+	body, err := json.Marshal(task)
 	if err != nil {
-			log.WithError(err)
-			continue
+		return err
 	}
 
-		ctx := context.Background()
-		if _, err = rnr.Run(ctx, cfg); err != nil {
-			log.WithError(err)
-			continue
-		}
-
-		log.Info("Processed task:", task.ID)
+	// Send out Delete request to delete task from queue
 	req, err := http.NewRequest(http.MethodDelete, url, bytes.NewBuffer(body))
 	if err != nil {
-			log.WithError(err)
+		return err
 	}
-
 	c := &http.Client{}
-		if _, err := c.Do(req); err != nil {
+	if resp, err := c.Do(req); err != nil {
+		return err
+	} else if resp.StatusCode != http.StatusAccepted {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+		return errors.New(string(body))
+	}
+	return nil
+}
+
+func runTask(task *models.Task) error {
+	// Set up runner and process task
+	cfg := getCfg(task)
+	ctx := context.Background()
+	rnr, err := New(NewMetricLogger())
+	if err != nil {
+		return err
+	}
+	_, err = rnr.Run(ctx, cfg)
+	return err
+}
+
+// RunAsyncRunner pulls tasks off a queue and processes them
+func RunAsyncRunner(mqAdr string) {
+	url := fmt.Sprintf("http://%s/tasks", mqAdr)
+	for {
+		task, err := getTask(url)
+		if err != nil {
+			log.WithError(err)
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		log.Info("Picked up task:", task.ID)
+
+		// Process Task
+		if err := runTask(task); err != nil {
 			log.WithError(err)
 			continue
 		}
+		log.Info("Processed task:", task.ID)
 
+		// Delete task from queue
+		if err := deleteTask(url, task); err != nil {
+			log.WithError(err)
+		} else {
 			log.Info("Deleted task:", task.ID)
 		}
+	}
 }

api/runner/async_runner_test.go

@@ -0,0 +1,144 @@
+package runner
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"math/rand"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/Sirupsen/logrus"
+	"github.com/gin-gonic/gin"
+	"github.com/iron-io/functions/api/models"
+	"github.com/iron-io/functions/api/mqs"
+)
+
+func getMockTask() models.Task {
+	priority := int32(0)
+	image := fmt.Sprintf("Image-%d", rand.Int31()%1000)
+	task := &models.Task{}
+	task.Image = &image
+	task.ID = fmt.Sprintf("ID-%d", rand.Int31()%1000)
+	task.RouteName = fmt.Sprintf("RouteName-%d", rand.Int31()%1000)
+	task.Priority = &priority
+	return *task
+}
+
+func getTestServer(mockTasks []*models.Task) *httptest.Server {
+	mq, err := mqs.New("memory://test")
+	if err != nil {
+		panic(err)
+	}
+
+	for _, mt := range mockTasks {
+		mq.Push(mt)
+	}
+
+	getHandler := func(c *gin.Context) {
+		task, err := mq.Reserve()
+		if err != nil {
+			logrus.WithError(err)
+			c.JSON(http.StatusInternalServerError, err)
+			return
+		}
+		c.JSON(http.StatusAccepted, task)
+	}
+
+	delHandler := func(c *gin.Context) {
+		body, err := ioutil.ReadAll(c.Request.Body)
+		if err != nil {
+			logrus.WithError(err)
+			c.JSON(http.StatusInternalServerError, err.Error())
+			return
+		}
+		var task models.Task
+		if err = json.Unmarshal(body, &task); err != nil {
+			logrus.WithError(err)
+			c.JSON(http.StatusInternalServerError, err.Error())
+			return
+		}
+
+		if err := mq.Delete(&task); err != nil {
+			logrus.WithError(err)
+			c.JSON(http.StatusInternalServerError, err.Error())
+			return
+		}
+		c.JSON(http.StatusAccepted, task)
+	}
+
+	r := gin.Default()
+	r.GET("/tasks", getHandler)
+	r.DELETE("/tasks", delHandler)
+	return httptest.NewServer(r)
+}
+
+func TestGetTask(t *testing.T) {
+	mockTask := getMockTask()
+
+	ts := getTestServer([]*models.Task{&mockTask})
+	defer ts.Close()
+
+	url := ts.URL + "/tasks"
+	task, err := getTask(url)
+	if err != nil {
+		t.Error("expected no error, got", err)
+	}
+	if task.ID != mockTask.ID {
+		t.Errorf("expected task ID '%s', got '%s'", task.ID, mockTask.ID)
+	}
+}
+
+func TestGetTaskError(t *testing.T) {
+	tests := []map[string]interface{}{
+		map[string]interface{}{
+			"url":   "/invalid",
+			"task":  getMockTask(),
+			"error": "invalid character 'p' after top-level value",
+		},
+	}
+
+	var tasks []*models.Task
+	for _, v := range tests {
+		task := v["task"].(models.Task)
+		tasks = append(tasks, &task)
+	}
+
+	ts := getTestServer(tasks)
+	defer ts.Close()
+
+	for i, test := range tests {
+		url := ts.URL + test["url"].(string)
+		_, err := getTask(url)
+		if err == nil {
+			t.Errorf("expected error '%s'", test["error"].(string))
+		}
+		if err.Error() != test["error"].(string) {
+			t.Errorf("test %d: expected error '%s', got '%s'", i, test["error"].(string), err)
+		}
+	}
+}
+
+func TestDeleteTask(t *testing.T) {
+	mockTask := getMockTask()
+
+	ts := getTestServer([]*models.Task{&mockTask})
+	defer ts.Close()
+
+	url := ts.URL + "/tasks"
+	err := deleteTask(url, &mockTask)
+	if err == nil {
+		t.Error("expected error 'Not reserver', got", err)
+	}
+
+	_, err = getTask(url)
+	if err != nil {
+		t.Error("expected no error, got", err)
+	}
+
+	err = deleteTask(url, &mockTask)
+	if err != nil {
+		t.Error("expected no error, got", err)
+	}
+}

main.go

@@ -59,7 +59,7 @@ func main() {
 	}
 
 	for i := 0; i < nasync; i++ {
-		go runner.RunAsyncRunners(mqAdr)
+		go runner.RunAsyncRunner(mqAdr)
 	}
 
 	quit := make(chan bool)