Add initial version with just 'running' endpoint that expects everyth…

…ing in the body

.gitignore

@@ -1 +1,2 @@
 job-status-listener
+job-status-listener.yaml

Gopkg.toml

@@ -31,3 +31,7 @@
 [prune]
   go-tests = true
   unused-packages = true
+
+[[constraint]]
+  branch = "master"
+  name = "github.com/cyverse-de/messaging"

main.go

@@ -1,15 +1,19 @@
 package main
 
 import (
+	"encoding/json"
 	_ "expvar"
 	"flag"
+	"net"
 	"net/http"
 
 	"github.com/cyverse-de/configurate"
 	"github.com/spf13/viper"
 
 	"github.com/gorilla/mux"
 	"github.com/sirupsen/logrus"
+
+	"github.com/cyverse-de/messaging"
 )
 
 var log = logrus.WithFields(logrus.Fields{
@@ -21,8 +25,75 @@ var log = logrus.WithFields(logrus.Fields{
 var (
 	cfgPath = flag.String("config", "", "Path to the configuration file.")
 	cfg     *viper.Viper
+
+	client *messaging.Client
 )
 
+// JobUpdatePublisher is the interface for types that need to publish a job
+// update.
+type JobUpdatePublisher interface {
+	PublishJobUpdate(m *messaging.UpdateMessage) error
+}
+
+func running(client JobUpdatePublisher, job *messaging.JobDetails, hostname string, msg string) (*messaging.UpdateMessage, error) {
+	parsedIP := net.ParseIP(hostname)
+	if parsedIP == nil {
+		_, err := net.LookupIP(hostname)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	updateMessage := &messaging.UpdateMessage{
+		Job:     *job,
+		State:   messaging.RunningState,
+		Message: msg,
+		Sender:  hostname,
+	}
+
+	err := client.PublishJobUpdate(updateMessage)
+	if err != nil {
+		log.Error(err)
+		return nil, err
+	}
+	log.Info(msg)
+	return updateMessage, nil
+}
+
+type MessagePost struct {
+	Hostname string
+	Message  string
+	Job      *messaging.JobDetails
+}
+
+func postRunning(w http.ResponseWriter, r *http.Request) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	out := json.NewEncoder(w)
+
+	var updateMessage MessagePost
+
+	err := json.NewDecoder(r.Body).Decode(&updateMessage)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		log.Error(err)
+		out.Encode(map[string]string{
+			"error": err.Error(),
+		})
+		return
+	}
+
+	msg, err := running(client, updateMessage.Job, updateMessage.Hostname, updateMessage.Message)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		log.Error(err)
+		out.Encode(map[string]string{
+			"error": err.Error(),
+		})
+		return
+	}
+	out.Encode(msg)
+}
+
 func init() {
 	flag.Parse()
 	logrus.SetFormatter(&logrus.JSONFormatter{})
@@ -39,6 +110,7 @@ func loadConfig(cfgPath string) {
 func newRouter() *mux.Router {
 	r := mux.NewRouter()
 	r.Handle("/debug/vars", http.DefaultServeMux)
+	r.Path("/running").Methods("POST").HandlerFunc(postRunning)
 
 	return r
 }
@@ -47,6 +119,17 @@ func main() {
 	log.Info("Starting up the job-status-listener service.")
 	loadConfig(*cfgPath)
 
+	uri := cfg.GetString("amqp.uri")
+	exchange := cfg.GetString("amqp.exchange.name")
+	var err error
+	client, err = messaging.NewClient(uri, true)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer client.Close()
+
+	client.SetupPublishing(exchange)
+
 	r := newRouter()
 
 	listenPortSpec := ":" + "60000"