Set message content-type based on the trigger.Spec.ContentType (#279)

fission/main.go

@@ -84,10 +84,11 @@ func main() {
 	mqtMQTypeFlag := cli.StringFlag{Name: "mqtype", Usage: "Message queue type, e.g. nats-streaming  (optional; uses \"nats-streaming\" if unspecified)"}
 	mqtTopicFlag := cli.StringFlag{Name: "topic", Usage: "Message queue Topic the trigger listens on"}
 	mqtRespTopicFlag := cli.StringFlag{Name: "resptopic", Usage: "Topic that the function response is sent on (optional; response discarded if unspecified)"}
+	mqtMsgContentType := cli.StringFlag{Name: "contenttype, c", Usage: "Content type of messages that publish to the topic (optional; uses \"application/json\" if unspecified)"}
 	mqtSubcommands := []cli.Command{
-		{Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag}, Action: mqtCreate},
+		{Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag, mqtMsgContentType}, Action: mqtCreate},
 		{Name: "get", Usage: "Get message queue trigger", Flags: []cli.Flag{}, Action: mqtGet},
-		{Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtTopicFlag, mqtRespTopicFlag, mqtFnNameFlag}, Action: mqtUpdate},
+		{Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtTopicFlag, mqtRespTopicFlag, mqtFnNameFlag, mqtMsgContentType}, Action: mqtUpdate},
 		{Name: "delete", Usage: "Delete message queue trigger", Flags: []cli.Flag{mqtNameFlag}, Action: mqtDelete},
 		{Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag}, Action: mqtList},
 	}

fission/mqtrigger.go

@@ -65,6 +65,11 @@ func mqtCreate(c *cli.Context) error {
 		fatal("Listen topic should not equal to response topic")
 	}
 
+	contentType := c.String("contenttype")
+	if len(contentType) == 0 {
+		contentType = "application/json"
+	}
+
 	checkMQTopicAvailability(mqType, topic, respTopic)
 
 	mqt := tpr.Messagequeuetrigger{
@@ -80,6 +85,7 @@ func mqtCreate(c *cli.Context) error {
 			MessageQueueType: mqType,
 			Topic:            topic,
 			ResponseTopic:    respTopic,
+			ContentType:      contentType,
 		},
 	}
 
@@ -103,6 +109,7 @@ func mqtUpdate(c *cli.Context) error {
 	topic := c.String("topic")
 	respTopic := c.String("resptopic")
 	fnName := c.String("function")
+	contentType := c.String("contenttype")
 
 	mqt, err := client.MessageQueueTriggerGet(&api.ObjectMeta{
 		Name:      mqtName,
@@ -125,6 +132,10 @@ func mqtUpdate(c *cli.Context) error {
 		mqt.Spec.FunctionReference.Name = fnName
 		updated = true
 	}
+	if len(contentType) > 0 {
+		mqt.Spec.ContentType = contentType
+		updated = true
+	}
 
 	if !updated {
 		fatal("Nothing to update. Use --topic, --resptopic, or --function.")
@@ -162,11 +173,11 @@ func mqtList(c *cli.Context) error {
 
 	w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0)
 
-	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n",
+	fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
-		"NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC")
+		"NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "PUB_MSG_CONTENT_TYPE")
 	for _, mqt := range mqts {
-		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\n",
+		fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\n",
-			mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic)
+			mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ContentType)
 	}
 	w.Flush()
 

mqtrigger/messageQueue/nats.go

@@ -106,6 +106,7 @@ func msgHandler(nats *Nats, trigger *tpr.Messagequeuetrigger) func(*ns.Msg) {
 		headers := map[string]string{
 			"X-Fission-MQTrigger-Topic":     trigger.Spec.Topic,
 			"X-Fission-MQTrigger-RespTopic": trigger.Spec.ResponseTopic,
+			"Content-Type":                  trigger.Spec.ContentType,
 		}
 
 		// Create request

types.go

@@ -159,6 +159,7 @@ type (
 		MessageQueueType  string            `json:"messageQueueType"`
 		Topic             string            `json:"topic"`
 		ResponseTopic     string            `json:"respTopic,omitempty"`
+		ContentType       string            `json:"contentType"`
 	}
 
 	// TimeTrigger invokes the specific function at a time or