Merge pull request #15 from uc-cdis/feat/dispatch_job_api

Feat/dispatch job api

Author: giangbui

handlers/handler.go

@@ -34,6 +34,11 @@ type JobConfig struct {
 	RequestMem  string      `request_mem`
 }
 
+type RetryMessage struct {
+	Bucket string `bucket`
+	Key    string `key`
+}
+
 // NewSQSHandler creates new SQSHandler instance
 func NewSQSHandler(queueURL string) *SQSHandler {
 	sqsHandler := new(SQSHandler)
@@ -56,7 +61,7 @@ func (handler *SQSHandler) StartServer() error {
 
 	glog.Info("Starting a new server...")
 	handler.Server = mq.NewServer(handler.QueueURL, mq.HandlerFunc(func(m *mq.Message) error {
-		return handler.HandleSQSMessage(m)
+		return handler.HandleSQSMessage(aws.StringValue(m.SQSMessage.Body))
 	}), mq.WithClient(newClient))
 	handler.Server.Start()
 	glog.Info("The server is started")
@@ -129,10 +134,10 @@ The format of a SQS message body:
 }
 */
 
-func getObjectsFromSQSMessage(m *mq.Message) []string {
+func getObjectsFromSQSMessage(msgBody string) []string {
 	objectPaths := make([]string, 0)
 	mapping := make(map[string][]interface{})
-	msgBody := aws.StringValue(m.SQSMessage.Body)
+	//msgBody := aws.StringValue(m.SQSMessage.Body)
 
 	msgBodyInf, err := GetValueFromJSON([]byte(msgBody), []string{"Message"})
 	if err != nil {
@@ -189,8 +194,9 @@ to the queue and retry later (handled by `md` library). That makes sure
 the message is properly handle before it actually deleted
 
 */
-func (handler *SQSHandler) HandleSQSMessage(m *mq.Message) error {
-	objectPaths := getObjectsFromSQSMessage(m)
+func (handler *SQSHandler) HandleSQSMessage(jsonBody string) error {
+
+	objectPaths := getObjectsFromSQSMessage(jsonBody)
 
 	jobNameList := make([]string, 0)
 	for _, jobConfig := range handler.JobConfigs {
@@ -268,3 +274,19 @@ func (handler *SQSHandler) handleListJobConfigs() (string, error) {
 	}
 	return "[" + str + "]", nil
 }
+
+/*
+RetryCreateIndexingJob creates manually job
+*/
+func (handler *SQSHandler) RetryCreateIndexingJob(jsonBytes []byte) error {
+
+	retryMessage := RetryMessage{}
+	if err := json.Unmarshal(jsonBytes, &retryMessage); err != nil {
+		return err
+	}
+
+	str := fmt.Sprintf(`{
+		"Type" : "Notification",
+		"Message" : "{\"Records\":[{\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventName\":\"ObjectCreated:Put\",\"s3\":{\"s3SchemaVersion\":\"1.0\",\"bucket\":{\"name\":\"%s\"},\"object\":{\"key\":\"%s\"}}}]}"}`, retryMessage.Bucket, retryMessage.Key)
+	return handler.HandleSQSMessage(str)
+}

handlers/handler_api.go

@@ -11,6 +11,7 @@ func (handler *SQSHandler) RegisterSQSHandler() {
 	http.HandleFunc("/server", handler.ServiceHandler)
 	http.HandleFunc("/sqs", handler.SQSHandler)
 	http.HandleFunc("/jobConfig", handler.HandleJobConfig)
+	http.HandleFunc("/dispatchJob", handler.HandleDispatchJob)
 }
 
 // ServiceHandler handles stop/start/status the SQS querrying service
@@ -138,3 +139,27 @@ func (handler *SQSHandler) listJobConfigs(w http.ResponseWriter, r *http.Request
 	}
 	fmt.Fprintf(w, str)
 }
+
+// HandleDispatchJob dispatch an job
+func (handler *SQSHandler) HandleDispatchJob(w http.ResponseWriter, r *http.Request) {
+	if r.Method == "POST" {
+		handler.dispatchJob(w, r)
+	}
+}
+
+// addJob adds an job config
+func (handler *SQSHandler) dispatchJob(w http.ResponseWriter, r *http.Request) {
+	// Try to read the request body.
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		msg := fmt.Sprintf("failed to read request body; encountered error: %s", err)
+		http.Error(w, msg, http.StatusBadRequest)
+		return
+	}
+	if err = handler.RetryCreateIndexingJob(body); err != nil {
+		msg := fmt.Sprintf("failed to dispatch an job; encountered error: %s", err)
+		http.Error(w, msg, http.StatusBadRequest)
+		return
+	}
+	fmt.Fprintf(w, "Successfully dispatch a new job!")
+}