add template function to use kafka header value by key

docs/config.md

@@ -301,4 +301,4 @@ If unset, an empty key is assigned during translation from NATS to Kafka. If the
 
 For NATS Streaming connections channel is treated as the subject and durable name is treated as the reply to, so that reply key type will use the durable name as the key.
 
-Every destination, may it be channel, subject or topic may be set using a go template that gets the message as data. e.g `topic: "{{ .Subject }}"`. Two template functions are available: replace (`"{{ .Subject | replace \"event\" \"message\" }}`) and substring (`"{{ .Subject | substring 0 5 }}"`)
+Every destination, may it be channel, subject or topic may be set using a go template that gets the message as data. e.g `topic: "{{ .Subject }}"`. Three template functions are available: replace (`"{{ .Subject | replace \"event\" \"message\" }}`), substring (`"{{ .Subject | substring 0 5 }}"`) and getKafkaHeaderValue(`"{{ .Headers | getKafkaHeaderValue \"baz\"}}"`)

server/core/connector.go

@@ -550,6 +550,14 @@ func (conn *BridgeConnector) initDestTemplate(destTpl string) {
 			}
 			return s[start:end]
 		},
+		"getKafkaHeaderValue": func(key string, headers []sarama.RecordHeader) string {
+		   for _, header := range(headers) {
+			   if string(header.Key) == key {
+				   return string(header.Value)
+			   }
+		   }
+		   return ""
+	   },
 	}
 	var err error
 	conn.destTemplate, err = template.New("dest").Funcs(funcMap).Parse(destTpl)