http handler options extended (#16)

config/config.go

@@ -84,12 +84,23 @@ func createHandler(v *viper.Viper) handlers.Handler {
 	v.SetDefault("http.path", "/")
 	v.SetDefault("http.host", "localhost")
 	v.SetDefault("http.port", 80)
+	v.SetDefault("http.method", "POST")
+	v.SetDefault("http.host", "")
+	v.SetDefault("http.headers", map[string]string{})
 
 	httpEndpoint := v.GetString("http.endpoint")
 	if httpEndpoint == "" {
 		httpEndpoint = fmt.Sprintf("http://%v:%v%v", v.GetString("http.host"), v.GetString("http.port"), v.GetString("http.path"))
 	}
-	return handlers.NewHttpHandler(httpEndpoint)
+
+	options := &handlers.HttpHandlerOptions{
+		Endpoint: httpEndpoint,
+		Method:   v.GetString("http.method"),
+		Host:     v.GetString("http.host"),
+		Headers:  v.GetStringMapString("http.headers"),
+	}
+
+	return handlers.NewHttpHandler(options)
 }
 
 func createWorkers(v *viper.Viper, sources map[string]*v1.Source) []*pipe.Worker {

handlers/http.go

@@ -14,6 +14,13 @@ type httpHandler struct {
 	client *gentleman.Client
 }
 
+type HttpHandlerOptions struct {
+	Endpoint string
+	Method   string
+	Host     string
+	Headers  map[string]string
+}
+
 var handlerLogger = log.With().Str("scope", "Handler")
 
 func (h *httpHandler) Handle(ctx *v1.RequestContext, message v1.Message) (*v1.RawMessage, HandlerError) {
@@ -34,11 +41,22 @@ func (h *httpHandler) Handle(ctx *v1.RequestContext, message v1.Message) (*v1.Ra
 	}, nil
 }
 
-func NewHttpHandler(endpoint string) Handler {
+func NewHttpHandler(options *HttpHandlerOptions) Handler {
 	client := gentleman.New().
-		URL(endpoint).
+		URL(options.Endpoint).
+		Method(options.Method).
 		Use(timeout.Request(2 * time.Minute))
 
+	if options.Host != "" {
+		client.AddHeader("Host", options.Host)
+	}
+
+	if options.Headers != nil {
+		for header, value := range options.Headers {
+			client.AddHeader(header, value)
+		}
+	}
+
 	return &httpHandler{
 		client,
 	}

integration/docker/pipes/multi.yaml

@@ -18,4 +18,6 @@ pipes:
   - source: azure-error
     handler:
       http:
-        path: "/"
+        headers:
+          x-client-id: testing
+        path: "/?require-header=X-Client-Id,testing"

integration/docker/worker/main.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 
@@ -38,6 +39,16 @@ func main() {
 				return
 			}
 		}
+
+		requireHeader := r.URL.Query().Get("require-header")
+		if requireHeader != "" {
+			header := strings.Split(requireHeader, ",")
+			if r.Header.Get(header[0]) != header[1] {
+				w.WriteHeader(500)
+				return
+			}
+		}
+
 		incremnt := !(r.URL.Query().Get("no-increment") == "true")
 		w.WriteHeader(200)
 		p := t