override default CORS handler for httprouter

cmd/semaphore/config/arguments.go

@@ -61,7 +61,19 @@ func ConstructArguments(params *Semaphore) ([]config.Option, error) {
 	}
 
 	if params.HTTP.Address != "" {
-		arguments = append(arguments, semaphore.WithListener(http.NewListener(params.HTTP.Address, specs.Options{})))
+		arguments = append(
+			arguments,
+			semaphore.WithListener(
+				http.NewListener(
+					params.HTTP.Address,
+					http.WithOrigins(params.HTTP.Origin),
+					http.WithReadTimeout(params.HTTP.ReadTimeout),
+					http.WithWriteTimeout(params.HTTP.WriteTimeout),
+					http.WithKeyFile(params.HTTP.KeyFile),
+					http.WithCertFile(params.HTTP.CertFile),
+				),
+			),
+		)
 	}
 
 	if params.GraphQL.Address != "" {

cmd/semaphore/config/config.go

@@ -26,21 +26,15 @@ func Parse(options *hcl.Options, target *Semaphore) {
 	}
 
 	if options.GraphQL != nil && target.GraphQL.Address == "" {
-		target.GraphQL = GraphQL{
-			Address: options.GraphQL.Address,
-		}
+		target.GraphQL = GraphQL(*options.GraphQL)
 	}
 
 	if options.HTTP != nil && target.HTTP.Address == "" {
-		target.HTTP = HTTP{
-			Address: options.HTTP.Address,
-		}
+		target.HTTP = HTTP(*options.HTTP)
 	}
 
 	if options.GRPC != nil && target.GRPC.Address == "" {
-		target.GRPC = GRPC{
-			Address: options.GRPC.Address,
-		}
+		target.GRPC = GRPC(*options.GRPC)
 	}
 
 	if options.Prometheus != nil && target.Prometheus.Address == "" {
@@ -68,7 +62,12 @@ type Prometheus struct {
 
 // HTTP configurations
 type HTTP struct {
-	Address string
+	Address      string
+	Origin       []string
+	CertFile     string
+	KeyFile      string
+	ReadTimeout  string
+	WriteTimeout string
 }
 
 // GRPC configurations

examples/custom-functions/main.go

@@ -24,7 +24,7 @@ func main() {
 
 	client, err := semaphore.New(
 		semaphore.WithLogLevel("*", "debug"),
-		semaphore.WithListener(http.NewListener(":8080", specs.Options{})),
+		semaphore.WithListener(http.NewListener(":8080")),
 		semaphore.WithFlows(hcl.FlowsResolver("./*.hcl")),
 		semaphore.WithEndpoints(hcl.EndpointsResolver("./*.hcl")),
 		semaphore.WithSchema(protobuffers.SchemaResolver([]string{"./proto"}, "./proto/*.proto")),

examples/http/config.hcl

@@ -11,4 +11,4 @@ services {
     select "com.semaphore.*" {
         host = "https://jsonplaceholder.typicode.com/"
     }
-}
+}

examples/http/flow.hcl

@@ -5,7 +5,9 @@ endpoint "FetchLatestProject" "http" {
 }
 
 flow "FetchLatestProject" {
-	input "com.semaphore.Query" {}
+	input "com.semaphore.Query" {
+		header = ["Authorization", "Timestamp"]
+	}
 
 	resource "query" {
 		request "com.semaphore.Todo" "Get" {

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/micro/go-micro/v2 v2.3.0
 	github.com/mitchellh/go-wordwrap v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.2.1
+	github.com/rs/cors v1.7.0
 	github.com/spf13/cobra v0.0.6
 	github.com/spf13/pflag v1.0.5 // indirect
 	github.com/zclconf/go-cty v1.3.1

go.sum

@@ -435,6 +435,8 @@ github.com/rainycape/memcache v0.0.0-20150622160815-1031fa0ce2f2/go.mod h1:7tZKc
 github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
+github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/russross/blackfriday v1.5.2 h1:HyvC0ARfnZBqnXwABFeSZHpKvJHJJfPz81GNueLj0oo=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
 github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=

main_test.go

@@ -70,7 +70,7 @@ func TestNewClient(t *testing.T) {
 				WithServices(hcl.ServicesResolver(file.Path)),
 				WithSchema(mock.SchemaResolver(schema)),
 				WithCodec(json.NewConstructor()),
-				WithListener(http.NewListener(":0", nil)),
+				WithListener(http.NewListener(":0")),
 				WithCaller(http.NewCaller()),
 				WithLogLevel("*", "debug"),
 			)
@@ -162,7 +162,7 @@ func TestServe(t *testing.T) {
 				WithServices(hcl.ServicesResolver(file.Path)),
 				WithSchema(mock.SchemaResolver(schema)),
 				WithCodec(json.NewConstructor()),
-				WithListener(http.NewListener(":0", nil)),
+				WithListener(http.NewListener(":0")),
 				WithCaller(http.NewCaller()),
 				WithLogLevel("*", "debug"),
 			)
@@ -215,7 +215,7 @@ func TestErrServe(t *testing.T) {
 				WithServices(hcl.ServicesResolver(file.Path)),
 				WithSchema(mock.SchemaResolver(schema)),
 				WithCodec(json.NewConstructor()),
-				WithListener(http.NewListener(listener.Addr().String(), nil)),
+				WithListener(http.NewListener(listener.Addr().String())),
 				WithCaller(http.NewCaller()),
 				WithLogLevel("*", "debug"),
 			)

pkg/providers/hcl/intermediate.go

@@ -28,7 +28,12 @@ type GraphQL struct {
 
 // HTTP represent the HTTP option definitions
 type HTTP struct {
-	Address string `hcl:"address"`
+	Address      string   `hcl:"address"`
+	Origin       []string `hcl:"origin,optional"`
+	CertFile     string   `hcl:"cert_file,optional"`
+	KeyFile      string   `hcl:"key_file,optional"`
+	ReadTimeout  string   `hcl:"read_timeout,optional"`
+	WriteTimeout string   `hcl:"write_timeout,optional"`
 }
 
 // GRPC represent the gRPC option definitions

pkg/providers/openapi3/openapi_test.go

@@ -49,7 +49,7 @@ func TestOpenAPI3Generation(t *testing.T) {
 				semaphore.WithServices(hcl.ServicesResolver(file.Path)),
 				semaphore.WithEndpoints(hcl.EndpointsResolver(file.Path)),
 				semaphore.WithCodec(json.NewConstructor()),
-				semaphore.WithListener(http.NewListener(":0", nil)),
+				semaphore.WithListener(http.NewListener(":0")),
 				semaphore.WithListener(grpc.NewListener(":0", nil)),
 				semaphore.WithCaller(http.NewCaller()),
 			}

pkg/transport/http/http_test.go

@@ -48,6 +48,10 @@ func NewCallerFunc(fn func(context.Context, references.Store) error) flow.Call {
 
 func NewSimpleMockSpecs() *specs.ParameterMap {
 	return &specs.ParameterMap{
+		Header: specs.Header{
+			"Authorization": &specs.Property{},
+			"Timestamp":     &specs.Property{},
+		},
 		Property: &specs.Property{
 			Type:  types.Message,
 			Label: labels.Optional,

pkg/transport/http/listener.go

@@ -14,20 +14,19 @@ import (
 	"github.com/jexia/semaphore/pkg/codec"
 	"github.com/jexia/semaphore/pkg/codec/metadata"
 	"github.com/jexia/semaphore/pkg/core/trace"
-	"github.com/jexia/semaphore/pkg/specs"
 	"github.com/jexia/semaphore/pkg/specs/template"
 	"github.com/jexia/semaphore/pkg/transport"
 	"github.com/julienschmidt/httprouter"
 	"go.uber.org/zap"
 )
 
 // NewListener constructs a new listener for the given addr
-func NewListener(addr string, opts specs.Options) transport.NewListener {
+func NewListener(addr string, options ...ListenerOption) transport.NewListener {
 	return func(parent *broker.Context) transport.Listener {
 		module := broker.WithModule(parent, "listener", "http")
 		ctx := logger.WithLogger(logger.WithFields(module, zap.String("listener", "http")))
 
-		options, err := ParseListenerOptions(opts)
+		options, err := NewListenerOptions(options...)
 		if err != nil {
 			logger.Error(ctx, "unable to parse HTTP listener options, unexpected error", zap.Error(err))
 		}
@@ -37,8 +36,8 @@ func NewListener(addr string, opts specs.Options) transport.NewListener {
 			options: options,
 			server: &http.Server{
 				Addr:         addr,
-				ReadTimeout:  options.ReadTimeout,
-				WriteTimeout: options.WriteTimeout,
+				ReadTimeout:  options.readTimeout,
+				WriteTimeout: options.writeTimeout,
 			},
 		}
 	}
@@ -51,28 +50,28 @@ type Listener struct {
 	server  *http.Server
 	mutex   sync.RWMutex
 	router  http.Handler
-	headers string
 }
 
 // Name returns the name of the given listener
-func (listener *Listener) Name() string {
-	return "http"
-}
+func (listener *Listener) Name() string { return "http" }
 
 // Serve opens the HTTP listener and calls the given handler function on reach request
 func (listener *Listener) Serve() (err error) {
 	logger.Info(listener.ctx, "serving HTTP listener", zap.String("addr", listener.server.Addr))
 
-	listener.server.Handler = listener.HandleCors(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		listener.mutex.RLock()
-		if listener.router != nil {
-			listener.router.ServeHTTP(w, r)
-		}
-		listener.mutex.RUnlock()
-	}))
+	listener.server.Handler = http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			listener.mutex.RLock()
+			defer listener.mutex.RUnlock()
+
+			if listener.router != nil {
+				listener.router.ServeHTTP(w, r)
+			}
+		},
+	)
 
-	if listener.options.CertFile != "" && listener.options.KeyFile != "" {
-		err = listener.server.ListenAndServeTLS(listener.options.CertFile, listener.options.KeyFile)
+	if listener.options.certFile != "" && listener.options.keyFile != "" {
+		err = listener.server.ListenAndServeTLS(listener.options.certFile, listener.options.keyFile)
 	} else {
 		err = listener.server.ListenAndServe()
 	}
@@ -88,40 +87,42 @@ func (listener *Listener) Serve() (err error) {
 func (listener *Listener) Handle(ctx *broker.Context, endpoints []*transport.Endpoint, codecs map[string]codec.Constructor) error {
 	logger.Info(listener.ctx, "HTTP listener received new endpoints")
 
-	router := httprouter.New()
-	headers := map[string]struct{}{}
+	var (
+		router  = httprouter.New()
+		headers = make(UniqueStringItems)
+		methods = make(UniqueStringItems)
+	)
 
 	for _, endpoint := range endpoints {
+
 		options, err := ParseEndpointOptions(endpoint.Options)
 		if err != nil {
 			return fmt.Errorf("endpoint %s: %s", endpoint.Flow, err)
 		}
 
+		methods.Add(options.Method)
+
 		ctx := logger.WithFields(ctx, zap.String("endpoint", options.Endpoint), zap.String("method", options.Method))
 		handle, err := NewHandle(ctx, endpoint, options, codecs)
 		if err != nil {
 			return err
 		}
 
-		if handle.Request != nil {
-			if handle.Request.Meta != nil {
-				for header := range handle.Request.Meta.Params {
-					headers[header] = struct{}{}
+		if endpoint.Request != nil {
+			if endpoint.Request.Meta != nil {
+				for header := range endpoint.Request.Meta.Params {
+					headers.Add(header)
 				}
 			}
 		}
 
 		router.Handle(options.Method, options.Endpoint, handle.HTTPFunc)
 	}
 
-	list := make([]string, 0, len(headers))
-	for header := range headers {
-		list = append(list, header)
-	}
+	router.GlobalOPTIONS = OptionsHandler(listener.options.origins, headers.Get(), methods.Get())
 
 	listener.mutex.Lock()
 	listener.router = router
-	listener.headers = strings.Join(list, ", ")
 	listener.mutex.Unlock()
 
 	return nil
@@ -133,29 +134,6 @@ func (listener *Listener) Close() error {
 	return listener.server.Close()
 }
 
-// HandleCors handles the defining of cors headers for the incoming HTTP request
-func (listener *Listener) HandleCors(h http.Handler) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		headers := w.Header()
-		method := r.Header.Get("Access-Control-Request-Method")
-
-		headers.Add("Vary", "Origin")
-		headers.Add("Vary", "Access-Control-Request-Method")
-		headers.Add("Vary", "Access-Control-Request-Headers")
-
-		headers.Set("Access-Control-Allow-Origin", "*")
-		headers.Set("Access-Control-Allow-Headers", "*")
-		headers.Set("Access-Control-Allow-Methods", strings.ToUpper(method))
-
-		if r.Method != http.MethodOptions || r.Header.Get("Access-Control-Request-Method") == "" {
-			h.ServeHTTP(w, r)
-			return
-		}
-
-		w.WriteHeader(http.StatusOK)
-	})
-}
-
 // NewHandle constructs a new handle function for the given endpoint to the given flow
 func NewHandle(ctx *broker.Context, endpoint *transport.Endpoint, options *EndpointOptions, constructors map[string]codec.Constructor) (*Handle, error) {
 	if constructors == nil {